neva 0.5.1

//! Utilities for the MCP client

use crate::error::{Error, ErrorCode};
use crate::shared;
use crate::transport::Transport;
use crate::types::Root;
use crate::types::sampling::{CreateMessageRequestParams, CreateMessageResult, SamplingHandler};
use crate::types::{
    CallToolRequestParams, CallToolResponse, GetPromptRequestParams, GetPromptResult,
    Implementation, ListPromptsRequestParams, ListPromptsResult,
    ListResourceTemplatesRequestParams, ListResourceTemplatesResult, ListResourcesRequestParams,
    ListResourcesResult, ListToolsRequestParams, ListToolsResult, MessageEnvelope,
    ReadResourceRequestParams, ReadResourceResult, Request, RequestId, RequestParamsMeta, Response,
    ServerCapabilities, Uri,
    cursor::Cursor,
    elicitation::{ElicitRequestParams, ElicitResult, ElicitationHandler},
    notification::Notification,
    resource::{SubscribeRequestParams, UnsubscribeRequestParams},
};
use crate::types::{ClientCapabilities, InitializeRequestParams, InitializeResult};
#[cfg(not(feature = "legacy-spec"))]
use crate::types::{SubscriptionFilter, SubscriptionsListenRequestParams};
use handler::RequestHandler;
use options::McpOptions;
use serde::Serialize;
use std::fmt::{Debug, Formatter};
use std::{future::Future, sync::Arc};
use tokio_util::sync::CancellationToken;

#[cfg(feature = "tasks")]
#[cfg(feature = "legacy-spec")]
use serde::de::DeserializeOwned;

#[cfg(feature = "tasks")]
use crate::types::{CancelTaskRequestParams, GetTaskRequestParams, TaskMetadata};
#[cfg(all(feature = "tasks", not(feature = "legacy-spec")))]
use crate::types::{DetailedTask, UpdateTaskRequestParams};
#[cfg(all(feature = "tasks", feature = "legacy-spec"))]
use crate::types::{
    GetTaskPayloadRequestParams, ListTasksRequestParams, ListTasksResult, Task, TaskPayload,
};

pub mod batch;
mod handler;
mod notification_handler;
pub mod options;
pub mod subscribe;
#[cfg(not(feature = "legacy-spec"))]
pub mod subscription;
#[cfg(feature = "tasks")]
pub mod task;

pub use batch::BatchBuilder;
#[cfg(not(feature = "legacy-spec"))]
pub use subscription::{Subscription, SubscriptionEnd};
#[cfg(feature = "tasks")]
pub use task::TaskBuilder;

/// Represents an MCP client app
pub struct Client {
    /// MCP client options.
    options: McpOptions,

    /// Capabilities supported by the connected server.
    server_capabilities: Option<ServerCapabilities>,

    /// Implementation information of the connected server.
    server_info: Option<Implementation>,

    /// A [`CancellationToken`] that cancels transport background processes.
    cancellation_token: Option<CancellationToken>,

    /// Request handler
    handler: Option<RequestHandler>,
}

impl Debug for Client {
    #[inline]
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Client")
            .field("options", &self.options)
            .field("server_capabilities", &self.server_capabilities)
            .field("server_info", &self.server_info)
            .finish()
    }
}

impl Default for Client {
    #[inline]
    fn default() -> Self {
        Self::new()
    }
}

impl Client {
    /// Initializes a new client app
    pub fn new() -> Self {
        Self {
            options: McpOptions::default(),
            server_capabilities: None,
            server_info: None,
            cancellation_token: None,
            handler: None,
        }
    }

    /// Configure MCP client options
    pub fn with_options<F>(mut self, config: F) -> Self
    where
        F: FnOnce(McpOptions) -> McpOptions,
    {
        self.options = config(self.options);
        self
    }

    /// Adds a new Root
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// # use neva::error::Error;
    ///
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Error> {
    /// let mut client = Client::new();
    /// client.add_root("file:///home/user/projects/my_project", "My Project");
    /// # client.disconnect().await
    /// # }
    /// ```
    #[deprecated(
        note = "Roots are deprecated in MCP 2026-07-28: the capability-driven `roots/list` request is gone and the ability is re-homed onto MRTR -- see `Context::list_roots`. Under MCP 2026-07-28 this configures what the client answers MRTR `roots/list` input requests with."
    )]
    #[allow(deprecated)]
    pub fn add_root(&mut self, uri: impl Into<Uri>, name: impl Into<String>) -> &mut Self {
        self.options.add_root(Root::new(uri, name));
        self.publish_roots_changed();
        self
    }

    /// Adds multiple new Roots.
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// # use neva::error::Error;
    ///
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Error> {
    /// let mut client = Client::new();
    /// client.add_roots([
    ///     ("file:///home/user/projects/my_project", "My Project"),
    ///     ("file:///home/user/projects/another_project", "My Another Project")
    /// ]);
    /// # client.disconnect().await
    /// # }
    /// ```
    #[deprecated(
        note = "Roots are deprecated in MCP 2026-07-28: the capability-driven `roots/list` request is gone and the ability is re-homed onto MRTR -- see `Context::list_roots`. Under MCP 2026-07-28 this configures what the client answers MRTR `roots/list` input requests with."
    )]
    #[allow(deprecated)]
    pub fn add_roots<T, I>(&mut self, roots: I) -> &mut Self
    where
        T: Into<Root>,
        I: IntoIterator<Item = T>,
    {
        self.options.add_roots(roots);
        self.publish_roots_changed();
        self
    }

    /// Sends the "notifications/roots/list_changed" notification to the server
    #[deprecated(
        note = "Roots are deprecated in MCP 2026-07-28: the capability-driven `roots/list` request is gone and the ability is re-homed onto MRTR -- see `Context::list_roots`. Under MCP 2026-07-28 this configures what the client answers MRTR `roots/list` input requests with."
    )]
    pub fn publish_roots_changed(&mut self) {
        if let Some(handler) = self.handler.as_mut() {
            let roots = self.options.roots();
            handler.notify_roots_changed(roots);
        }
    }

    /// Registers a handler that will be running when a "sampling/createMessage" request is received
    #[deprecated(
        note = "Sampling is deprecated in MCP 2026-07-28: the capability-driven `sampling/createMessage` request is gone and the ability is re-homed onto MRTR -- see `Context::sample`. Under MCP 2026-07-28 this handler fulfils MRTR `sampling/createMessage` input requests."
    )]
    pub fn map_sampling<F, R>(&mut self, handler: F) -> &mut Self
    where
        F: Fn(CreateMessageRequestParams) -> R + Clone + Send + Sync + 'static,
        R: Future + Send,
        R::Output: Into<CreateMessageResult>,
    {
        let handler: SamplingHandler = make_handler(handler);
        self.options.add_sampling_handler(handler);
        self
    }

    /// Registers a handler that will be running when an "elicitation/create" request is received
    pub fn map_elicitation<F, R>(&mut self, handler: F) -> &mut Self
    where
        F: Fn(ElicitRequestParams) -> R + Clone + Send + Sync + 'static,
        R: Future + Send,
        R::Output: Into<ElicitResult>,
    {
        let handler: ElicitationHandler = make_handler(handler);
        self.options.add_elicitation_handler(handler);
        self
    }

    /// Connects the MCP client to the MCP server
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // call tools, read resources, etc.
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn connect(&mut self) -> Result<(), Error> {
        #[cfg(feature = "macros")]
        self.register_methods();

        let mut transport = self.options.transport();
        let token = transport.start();

        #[cfg(feature = "tracing")]
        self.register_tracing_notification_handlers();

        self.cancellation_token = Some(token.clone());
        self.handler = Some(RequestHandler::new(transport, &self.options, token));

        self.wait_for_shutdown_signal();
        self.init().await
    }

    /// Disconnects the MCP client from the MCP server
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // call tools, read resources, etc.
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn disconnect(mut self) -> Result<(), Error> {
        self.send_notification(crate::types::notification::commands::CANCELLED, None)
            .await?;
        if let Some(token) = self.cancellation_token {
            token.cancel();
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        Ok(())
    }

    /// The protocol version this client expects from the connected peer.
    ///
    /// Under MCP 2026-07-28 the 2026-07-28 expectation is pinned to
    /// [`crate::LATEST_PROTOCOL_VERSION`]: a `with_mcp_version` override only
    /// selects which legacy version the dual-mode fallback negotiates --
    /// it must never make `server/discover` reject a valid 2026-07-28 server.
    fn expected_protocol_ver(&self) -> &'static str {
        #[cfg(not(feature = "legacy-spec"))]
        {
            if self.is_legacy_peer() {
                self.options.legacy_protocol_ver()
            } else {
                crate::LATEST_PROTOCOL_VERSION
            }
        }
        #[cfg(feature = "legacy-spec")]
        {
            self.options.protocol_ver()
        }
    }

    /// Validates a server-reported protocol version against what this client
    /// negotiated, cancelling the transport on mismatch.
    fn validate_server_version(&mut self, server_ver: &str) -> Result<(), Error> {
        if !crate::PROTOCOL_VERSIONS.contains(&server_ver) {
            self.cancel_transport();
            return Err(Error::new(
                ErrorCode::InvalidRequest,
                format!("Unsupported server protocol version: {server_ver}"),
            ));
        }
        if server_ver != self.expected_protocol_ver() {
            self.cancel_transport();
            return Err(Error::new(
                ErrorCode::InvalidRequest,
                format!(
                    "Server protocol version mismatch: expected {}, got {server_ver}",
                    self.expected_protocol_ver()
                ),
            ));
        }
        Ok(())
    }

    /// Sends `initialize` request to an MCP server (legacy handshake).
    #[cfg(feature = "legacy-spec")]
    pub async fn init(&mut self) -> Result<(), Error> {
        self.legacy_init().await
    }

    /// The `initialize`/`initialized` handshake: the only handshake for
    /// the legacy build, the dual-mode fallback for the 2026-07-28 build.
    async fn legacy_init(&mut self) -> Result<(), Error> {
        #[cfg(feature = "legacy-spec")]
        let protocol_ver = self.options.protocol_ver().to_string();
        // The fallback negotiates the newest legacy version -- offering
        // the 2026-07-28 version to a server that just rejected `server/discover`
        // would only get refused again.
        #[cfg(not(feature = "legacy-spec"))]
        let protocol_ver = self.options.legacy_protocol_ver().to_string();

        let params = InitializeRequestParams {
            protocol_ver,
            client_info: Some(self.options.implementation.clone()),
            capabilities: Some(ClientCapabilities {
                roots: self.options.roots_capability(),
                sampling: self.options.sampling_capability(),
                elicitation: self.options.elicitation_capability(),
                #[cfg(feature = "tasks")]
                tasks: self.options.tasks_capability(),
                #[cfg(not(feature = "legacy-spec"))]
                extensions: None,
                experimental: None,
            }),
        };

        let req = Request::new(
            Some(RequestId::Uuid(uuid::Uuid::new_v4())),
            crate::commands::INIT,
            Some(params),
        );

        let resp = self.send_request(req).await?;

        let init_result = resp.into_result::<InitializeResult>()?;

        self.validate_server_version(init_result.protocol_ver.as_str())?;

        self.server_capabilities = Some(init_result.capabilities);
        self.server_info = Some(init_result.server_info);

        self.send_notification(crate::types::notification::commands::INITIALIZED, None)
            .await
    }

    /// Discovers server capabilities via `server/discover` (MCP 2026-07-28).
    ///
    /// Replaces the `initialize`/`initialized` handshake. No `initialized`
    /// notification is sent -- the transport is stateless.
    #[cfg(not(feature = "legacy-spec"))]
    pub async fn discover(&mut self) -> Result<(), Error> {
        let resp = self.send_request(Self::discover_request()).await?;
        let result = resp.into_result::<crate::types::DiscoverResult>()?;
        self.apply_discover(result)
    }

    /// Builds the `server/discover` request.
    #[cfg(not(feature = "legacy-spec"))]
    fn discover_request() -> Request {
        Request::new(
            Some(RequestId::Uuid(uuid::Uuid::new_v4())),
            crate::commands::DISCOVER,
            Some(crate::types::DiscoverRequestParams::default()),
        )
    }

    /// Applies a successful `server/discover` result: validates the
    /// reported protocol version and stores the capabilities.
    #[cfg(not(feature = "legacy-spec"))]
    fn apply_discover(&mut self, result: crate::types::DiscoverResult) -> Result<(), Error> {
        // Discovery advertises a *set*; the handshake succeeds when the version
        // this client speaks is among them.
        let expected = self.expected_protocol_ver();
        if !result.supported_versions.iter().any(|v| v == expected) {
            // `connect` has already started the transport; leaving it running
            // would park background HTTP/SSE tasks behind a client that never
            // completed its handshake.
            self.cancel_transport();
            return Err(Error::new(
                ErrorCode::UnsupportedProtocolVersion,
                format!(
                    "Server supports {:?} but the client speaks {expected}",
                    result.supported_versions
                ),
            ));
        }
        self.server_capabilities = Some(result.capabilities);
        // `serverInfo` left `DiscoverResult` in the final spec: servers now
        // report themselves in every result's `_meta`, so it is picked up from
        // there instead.
        Ok(())
    }

    /// The dual-mode handshake (issue #84): tries `server/discover`
    /// first and, when the server clearly doesn't speak the 2026-07-28 protocol,
    /// falls back to the legacy `initialize` handshake and marks the
    /// peer as legacy -- subsequent traffic uses legacy semantics
    /// (session header, SSE stream, no MRTR, no 2026-07-28 routing headers).
    ///
    /// Only **wire-phase** failures classify for the fallback: transport
    /// errors and the server's JSON-RPC *error* reply. Once the server
    /// answers `server/discover` successfully, the peer has committed to
    /// the 2026-07-28 protocol -- later local failures (a malformed result, an
    /// unsupported/mismatched `protocolVersion`) surface as real errors
    /// instead of a misleading fallback attempt on a transport that
    /// version validation may already have cancelled.
    #[cfg(not(feature = "legacy-spec"))]
    pub async fn init(&mut self) -> Result<(), Error> {
        let resp = match self.send_request(Self::discover_request()).await {
            Ok(resp) => resp,
            Err(err) if is_fallback_trigger(&err) => return self.fallback_init(&err).await,
            Err(err) => return Err(err),
        };

        let is_error_reply = matches!(resp, Response::Err(_));
        let result = match resp.into_result::<crate::types::DiscoverResult>() {
            Ok(result) => result,
            Err(err) if is_error_reply && is_fallback_trigger(&err) => {
                return self.fallback_init(&err).await;
            }
            Err(err) => return Err(err),
        };

        self.apply_discover(result)
    }

    /// Runs the legacy half of the dual-mode handshake after a
    /// classified `server/discover` rejection.
    #[cfg(not(feature = "legacy-spec"))]
    async fn fallback_init(&mut self, _err: &Error) -> Result<(), Error> {
        #[cfg(feature = "tracing")]
        tracing::info!(
            logger = "neva",
            "`server/discover` rejected ({_err}); falling back to `initialize`"
        );
        self.options.peer_mode.set_legacy();
        self.legacy_init().await
    }

    /// Whether the peer negotiated the legacy protocol through the
    /// dual-mode fallback.
    #[cfg(not(feature = "legacy-spec"))]
    fn is_legacy_peer(&self) -> bool {
        self.options.peer_mode.is_legacy()
    }

    /// Sends a ping to the MCP server
    ///
    /// Removed in MCP 2026-07-28; available only under `legacy-spec`.
    #[cfg(feature = "legacy-spec")]
    pub async fn ping(&mut self) -> Result<Response, Error> {
        self.command::<()>(crate::commands::PING, None).await
    }

    /// Sends a command to the MCP server
    ///
    /// # Example
    /// ```no_run
    /// use neva::prelude::*;
    ///
    /// #[derive(serde::Serialize)]
    /// struct MyCommandParams {
    ///     param: String,
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let params = MyCommandParams { param: "Hello MCP!".to_string() };
    ///     let tools = client.command("my-command", Some(params)).await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    #[inline]
    pub async fn command<T: Serialize>(
        &mut self,
        command: impl Into<String>,
        params: Option<T>,
    ) -> Result<Response, Error> {
        let id = self.generate_id()?;
        let request = Request::new(Some(id), command, params);
        self.send_request(request).await
    }

    /// Requests a list of tools that MCP server provides
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // Fetch all or initial list of tools if the MCP server provides pagination
    ///     let tools = client.list_tools(None).await?;
    ///     
    ///     // Fetch the next page of tools is any   
    ///     let tools = client.list_tools(tools.next_cursor).await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn list_tools(&mut self, cursor: Option<Cursor>) -> Result<ListToolsResult, Error> {
        // A cursor-less call starts the listing over, so it replaces what the
        // previous traversal registered rather than merging into it.
        #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
        let fresh = cursor.is_none();
        let params = ListToolsRequestParams { cursor };

        #[allow(unused_mut)]
        let mut result: ListToolsResult = self
            .command(crate::types::tool::commands::LIST, Some(params))
            .await?
            .into_result()?;

        #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
        self.register_param_headers(&mut result, fresh);

        Ok(result)
    }

    /// Runs a batched `tools/list` response through the same registry update a
    /// direct [`Self::list_tools`] performs, rewriting the response in place so
    /// the caller never sees a tool the client refuses to call.
    ///
    /// A batched listing is always a fresh traversal: [`BatchBuilder`] enqueues
    /// it without a cursor. A response that does not parse as a listing is left
    /// alone -- it is the caller's to interpret, and it registers nothing.
    #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
    pub(super) fn register_batched_tools(&mut self, resp: &mut Response) {
        let Response::Ok(ok) = resp else { return };
        let Ok(mut result) = serde_json::from_value::<ListToolsResult>(ok.result.clone()) else {
            return;
        };
        self.register_param_headers(&mut result, true);
        if let Ok(value) = serde_json::to_value(&result) {
            ok.result = value;
        }
    }

    /// Records each tool's `x-mcp-header` annotations and drops any tool whose
    /// annotations are invalid.
    ///
    /// The spec makes rejection per-tool on purpose: one malformed definition
    /// must not take the whole listing down, and must not be callable either --
    /// so the offending tool is removed from the result the caller sees.
    ///
    /// A refreshed listing replaces what the previous one registered, including
    /// replacing it with nothing: a server that drops an annotation -- or drops
    /// the whole tool -- must stop the client from mirroring that argument into
    /// a header, which a leftover registration would keep doing even though the
    /// current listing no longer designates it.
    ///
    /// `fresh` marks the first page of a traversal, which clears the registry;
    /// later pages accumulate onto it, since a tool absent from page two has
    /// not been withdrawn, only listed elsewhere.
    ///
    /// The name of a rejected tool is remembered as well, so that hiding it
    /// from the listing is not all that hiding it does -- see
    /// [`Self::blocked_tool_error`].
    #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
    fn register_param_headers(&mut self, result: &mut ListToolsResult, fresh: bool) {
        use crate::shared::param_headers;

        if fresh {
            self.options.param_headers.clear();
            self.options.rejected_tools.clear();
        }

        result.tools.retain(|tool| {
            self.options.param_headers.remove(&*tool.name);
            self.options.rejected_tools.remove(&*tool.name);
            let schema = match serde_json::to_value(&tool.input_schema) {
                Ok(schema) => schema,
                Err(_) => return true,
            };
            match param_headers::collect(&schema) {
                Ok(headers) => {
                    if !headers.is_empty() {
                        self.options
                            .param_headers
                            .insert(tool.name.to_string(), headers);
                    }
                    true
                }
                Err(_err) => {
                    #[cfg(feature = "tracing")]
                    tracing::warn!(logger = "neva", "Dropping tool `{}`: {_err}", tool.name);
                    self.options.rejected_tools.insert(tool.name.to_string());
                    false
                }
            }
        });
    }

    /// Refuses a `tools/call` naming a tool the current listing withdrew for a
    /// malformed `x-mcp-header` declaration.
    ///
    /// Dropping such a tool from `tools/list` is what the spec asks for, but on
    /// its own it only hides the name: a caller holding one from somewhere else
    /// -- hard-coded, cached, read off a log -- still reaches `call_tool`, and
    /// since the declaration never parsed there are no annotations to mirror,
    /// so the call would travel with none of the `Mcp-Param-*` headers it asked
    /// for. An intermediary would see a call it cannot route or police, which
    /// is the one outcome the annotation exists to prevent -- so the call is
    /// refused instead of quietly sent unannotated.
    ///
    /// Only tools this client has seen rejected are known; one it never listed
    /// cannot be recognized.
    ///
    /// Sits on the send seam, so every request pays for it -- an empty set is
    /// checked first precisely so that the requests it is not about (and, in a
    /// healthy connection, all of them) stop at a single branch.
    #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
    #[inline]
    fn blocked_tool_error(&self, req: &Request) -> Option<Error> {
        if self.options.rejected_tools.is_empty()
            || req.method.as_str() != crate::types::tool::commands::CALL
        {
            return None;
        }

        let name = req.params.as_ref()?.get("name")?.as_str()?;
        if !self.options.rejected_tools.contains(name) {
            return None;
        }

        Some(Error::new(
            ErrorCode::InvalidParams,
            format!(
                "Tool `{name}` was rejected for an invalid `x-mcp-header` declaration and cannot be called"
            ),
        ))
    }

    /// Requests a list of resources that MCP server provides
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // Fetch all or initial list of resources if the MCP server provides pagination
    ///     let resources = client.list_resources(None).await?;
    ///     
    ///     // Fetch the next page of resources is any   
    ///     let resources = client.list_resources(resources.next_cursor).await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn list_resources(
        &mut self,
        cursor: Option<Cursor>,
    ) -> Result<ListResourcesResult, Error> {
        let params = ListResourcesRequestParams { cursor };
        self.command(crate::types::resource::commands::LIST, Some(params))
            .await?
            .into_result()
    }

    /// Requests a list of resource templates that MCP server provides
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // Fetch all or initial list of resource templates if the MCP server provides pagination
    ///     let templates = client.list_resource_templates(None).await?;
    ///     
    ///     // Fetch the next page of resource templates is any   
    ///     let templates = client.list_resource_templates(templates.next_cursor).await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn list_resource_templates(
        &mut self,
        cursor: Option<Cursor>,
    ) -> Result<ListResourceTemplatesResult, Error> {
        let params = ListResourceTemplatesRequestParams { cursor };
        self.command(
            crate::types::resource::commands::TEMPLATES_LIST,
            Some(params),
        )
        .await?
        .into_result()
    }

    /// Requests a list of prompts that MCP server provides
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     // Fetch all or initial list of prompts if the MCP server provides pagination
    ///     let prompts = client.list_prompts(None).await?;
    ///     
    ///     // Fetch the next page of prompts templates is any   
    ///     let prompts = client.list_prompts(prompts.next_cursor).await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn list_prompts(
        &mut self,
        cursor: Option<Cursor>,
    ) -> Result<ListPromptsResult, Error> {
        let params = ListPromptsRequestParams { cursor };
        self.command(crate::types::prompt::commands::LIST, Some(params))
            .await?
            .into_result()
    }

    /// Calls a tool that MCP server supports
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let args = [("message", "Hello MCP!")]; // or let args = ("message", "Hello MCP!");
    ///     let result = client.call_tool("echo", args).await?;
    ///     // Do something with the result
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    ///
    /// # Structured output
    /// ```no_run
    /// use neva::prelude::*;
    ///
    /// #[json_schema(de)]
    /// struct Weather {
    ///     conditions: String,
    ///     temperature: f32,
    ///     humidity: f32,
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let tools = client.list_tools(None).await?;
    ///
    ///     // Get the tool by name
    ///     let tool: &Tool = tools.get("weather-forecast")
    ///         .expect("Weather forecast tool not found");
    ///
    ///     let args = ("location", "London");
    ///     let result = client.call_tool("weather-forecast", args).await?;
    ///
    ///     // Validate the output structure and deserialize the result
    ///     let weather: Weather = tool
    ///         .validate(&result)
    ///         .and_then(|res| res.as_json())?;
    ///     
    ///     // Do something with the result
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn call_tool<N, Args>(
        &mut self,
        name: N,
        args: Args,
    ) -> Result<CallToolResponse, Error>
    where
        N: Into<String>,
        Args: shared::IntoArgs,
    {
        let params = CallToolRequestParams {
            name: name.into(),
            meta: None,
            args: args.into_args(),
            #[cfg(feature = "tasks")]
            task: None,
        };

        self.call_tool_raw(params).await?.into_result()
    }

    /// Calls a task-augmented tool that MCP server supports
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let args = [("message", "Hello MCP!")]; // or let args = ("message", "Hello MCP!");
    ///     let result = client.call_tool_as_task("echo", args, None).await?;
    ///     // Do something with the result
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    ///
    /// # Structured output
    /// ```no_run
    /// use neva::prelude::*;
    ///
    /// #[json_schema(de)]
    /// struct Weather {
    ///     conditions: String,
    ///     temperature: f32,
    ///     humidity: f32,
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let tools = client.list_tools(None).await?;
    ///
    ///     // Get the tool by name
    ///     let tool: &Tool = tools.get("weather-forecast")
    ///         .expect("Weather forecast tool not found");
    ///
    ///     let args = ("location", "London");
    ///     let result = client.call_tool_as_task("weather-forecast", args, None).await?;
    ///
    ///     // Validate the output structure and deserialize the result
    ///     let weather: Weather = tool
    ///         .validate(&result)
    ///         .and_then(|res| res.as_json())?;
    ///     
    ///     // Do something with the result
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    #[cfg(feature = "tasks")]
    pub async fn call_tool_as_task<N, Args>(
        &mut self,
        name: N,
        args: Args,
        ttl: Option<usize>,
    ) -> Result<CallToolResponse, Error>
    where
        N: Into<String>,
        Args: shared::IntoArgs,
    {
        let builder = self.task();
        let builder = if let Some(t) = ttl {
            builder.with_ttl(t)
        } else {
            builder
        };
        builder.call_tool(name, args).await
    }

    /// Calls a tool
    #[inline]
    pub async fn call_tool_raw(
        &mut self,
        params: CallToolRequestParams,
    ) -> Result<Response, Error> {
        let id = self.generate_id()?;

        let request = Request::new(
            Some(id.clone()),
            crate::types::tool::commands::CALL,
            Some(params.with_meta(RequestParamsMeta::new(&id))),
        );

        self.send_request(request).await
    }

    /// Requests resource contents from MCP server
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let resource = client.read_resource("res://res_1").await?;
    ///     // Do something with the resource
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn read_resource(
        &mut self,
        uri: impl Into<Uri>,
    ) -> Result<ReadResourceResult, Error> {
        let id = self.generate_id()?;
        let request = Request::new(
            Some(id.clone()),
            crate::types::resource::commands::READ,
            Some(ReadResourceRequestParams {
                uri: uri.into(),
                meta: Some(RequestParamsMeta::new(&id)),
                #[cfg(feature = "server")]
                args: None,
            }),
        );

        self.send_request(request).await?.into_result()
    }

    /// Gets a prompt that MCP server provides
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///
    ///     client.connect().await?;
    ///
    ///     let args = [
    ///         ("temperature", "50"),
    ///         ("style", "anything")
    ///     ];
    ///     let prompt = client.get_prompt("complex_prompt", args).await?;
    ///     // Do something with the prompt
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub async fn get_prompt<N, Args>(
        &mut self,
        name: N,
        args: Args,
    ) -> Result<GetPromptResult, Error>
    where
        N: Into<String>,
        Args: shared::IntoArgs,
    {
        let id = self.generate_id()?;
        let request = Request::new(
            Some(id.clone()),
            crate::types::prompt::commands::GET,
            Some(GetPromptRequestParams {
                name: name.into(),
                meta: Some(RequestParamsMeta::new(&id)),
                args: args.into_args(),
            }),
        );

        self.send_request(request).await?.into_result()
    }

    /// Opens a long-lived notification subscription (MCP 2026-07-28).
    ///
    /// Sends `subscriptions/listen` and returns once the server has
    /// acknowledged the filter. Notifications delivered on the stream are
    /// dispatched to the handlers registered with [`Self::subscribe`] and its
    /// helpers ([`Self::on_tools_changed`] and friends), so those must be in
    /// place before listening -- and, for the capability-gated helpers, after
    /// [`Self::connect`], which is what discovers the capabilities they assert
    /// on.
    ///
    /// The returned [`Subscription`] carries the accepted filter -- the server
    /// silently drops types it does not advertise -- and ends the stream on
    /// [`Subscription::cancel`].
    ///
    /// # Example
    /// ```no_run
    /// use neva::{Client, error::Error, types::SubscriptionFilter};
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///     client.connect().await?;
    ///     client.on_tools_changed(|_| async { println!("tools changed"); });
    ///
    ///     let subscription = client
    ///         .listen(SubscriptionFilter::new().with_tools_changed())
    ///         .await?;
    ///
    ///     println!("accepted: {:?}", subscription.acknowledged());
    ///     Ok(())
    /// }
    /// ```
    #[cfg(not(feature = "legacy-spec"))]
    pub async fn listen(
        &mut self,
        notifications: SubscriptionFilter,
    ) -> Result<Subscription, Error> {
        if self.is_legacy_peer() {
            return Err(Error::new(
                ErrorCode::MethodNotFound,
                "Peer speaks the legacy protocol; use subscribe_to_resource instead",
            ));
        }

        let id = self.generate_id()?;
        let mut request = Request::new(
            Some(id.clone()),
            crate::types::subscription::commands::LISTEN,
            Some(SubscriptionsListenRequestParams::new(notifications.clone())),
        );
        self.apply_client_meta(&mut request, None, None);

        let handler = self
            .handler
            .as_mut()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?;

        // Watch for the acknowledgment before sending: it is the first thing
        // the server puts on the stream, and the receive loop drops one nobody
        // is waiting for.
        let ack = handler.watch_ack(&id, &notifications);
        let sender = handler.sender();
        let release = handler.subscription_release();

        // Armed before the send, not after it. `watch_ack` has already
        // registered the waiter and the pending state, `send_listen` takes the
        // untimed request slot before it awaits the transport, and that await
        // is a suspension point like any other: a caller who drops this future
        // (an outer `tokio::time::timeout`, a lost `select!` branch) runs none
        // of the branches below, and everything registered so far would be left
        // behind. Nothing between here and `watch_ack` awaits, so there is no
        // gap left to fall into.
        let guard =
            subscription::EstablishmentGuard::new(id.clone(), release.clone(), sender.clone());

        let mut response = match handler.send_listen(request).await {
            Ok(response) => response,
            // Never reached the wire, so there is no stream to cancel -- only
            // this client's own bookkeeping to drop.
            Err(err) => {
                guard.forget();
                return Err(err);
            }
        };

        // Race the acknowledgment against the request's own reply: a peer that
        // rejects the subscription outright -- `MethodNotFound`, an
        // authorization failure, invalid params -- answers instead of
        // acknowledging, and waiting only on the acknowledgment would sit out
        // the whole timeout and report that instead of the server's reason.
        let timeout = self.options.timeout;
        let established = tokio::select! {
            biased;
            acknowledged = ack => Ok(acknowledged),
            answered = &mut response => Err(match answered {
                Ok(shared::PendingResponse::Response(resp)) => match resp {
                    // An error reply is the server's own explanation; surface it.
                    Response::Err(err) => err.error.into(),
                    // A success reply this early is the graceful-close result
                    // for a subscription that never carried anything.
                    Response::Ok(_) => Error::new(
                        ErrorCode::InternalError,
                        "Subscription ended before it was acknowledged",
                    ),
                },
                Ok(shared::PendingResponse::Timeout) => {
                    Error::new(ErrorCode::Timeout, "Subscription was not acknowledged")
                }
                // The slot's sender was dropped: the receive loop released it
                // on its way out, so the transport is gone. That is a lost
                // connection, not a peer that would not acknowledge, and
                // callers act on the two differently.
                Err(_) => Error::new(ErrorCode::InternalError, "Connection closed"),
            }),
            _ = tokio::time::sleep(timeout) => Err(Error::new(
                ErrorCode::Timeout,
                "Subscription was not acknowledged",
            )),
        };

        let acknowledged = match established {
            Ok(Ok(filter)) => filter,
            // The waiter's sender was dropped: the receive loop is gone.
            Ok(Err(_)) => {
                guard.abandon().await;
                return Err(Error::new(ErrorCode::InternalError, "Connection closed"));
            }
            Err(err) => {
                guard.abandon().await;
                return Err(err);
            }
        };

        // The server may narrow the filter -- that is the whole point of the
        // acknowledgment -- but it must not widen it. Notifications are
        // dispatched to the client's own handlers with no per-subscription
        // filtering, so an acknowledgment claiming a category or URI this call
        // never asked for would deliver events outside the requested scope.
        if !acknowledged.is_subset_of(&notifications) {
            guard.abandon().await;
            return Err(Error::new(
                ErrorCode::InvalidRequest,
                "Server acknowledged a subscription broader than the one requested",
            ));
        }

        // The handle takes over from here.
        guard.disarm();

        Ok(Subscription::new(
            id,
            notifications,
            acknowledged,
            response,
            sender,
            release,
        ))
    }

    /// Subscribes to a resource on the server to receive notifications when it changes.
    ///
    /// Legacy only in effect: MCP 2026-07-28 folds per-resource subscriptions
    /// into the `listen` filter, so against a 2026-07-28 peer this fails and
    /// `listen` with a `resourceSubscriptions` entry is the way. The method
    /// stays available because the dual-mode fallback still reaches legacy
    /// peers.
    pub async fn subscribe_to_resource(&mut self, uri: impl Into<Uri>) -> Result<(), Error> {
        #[cfg(not(feature = "legacy-spec"))]
        if !self.is_legacy_peer() {
            return Err(Error::new(
                ErrorCode::MethodNotFound,
                "resources/subscribe is legacy-only; use listen with a resource filter",
            ));
        }
        if !self.is_resource_subscription_supported() {
            return Err(Error::new(
                ErrorCode::MethodNotFound,
                "Server does not support resource subscriptions",
            ));
        }

        let params = SubscribeRequestParams::from(uri);
        let resp = self
            .command(crate::types::resource::commands::SUBSCRIBE, Some(params))
            .await?;

        match resp {
            Response::Ok(_) => Ok(()),
            Response::Err(err) => Err(err.error.into()),
        }
    }

    /// Unsubscribes from a resource on the server to stop receiving notifications about its changes.
    ///
    /// Legacy only in effect; see [`Self::subscribe_to_resource`]. Under MCP
    /// 2026-07-28 a subscription ends with the stream that carries it
    /// (`Subscription::cancel`).
    pub async fn unsubscribe_from_resource(&mut self, uri: impl Into<Uri>) -> Result<(), Error> {
        #[cfg(not(feature = "legacy-spec"))]
        if !self.is_legacy_peer() {
            return Err(Error::new(
                ErrorCode::MethodNotFound,
                "resources/unsubscribe is legacy-only; cancel the subscription instead",
            ));
        }
        if !self.is_resource_subscription_supported() {
            return Err(Error::new(
                ErrorCode::MethodNotFound,
                "Server does not support resource subscriptions",
            ));
        }

        let params = UnsubscribeRequestParams::from(uri);
        let resp = self
            .command(crate::types::resource::commands::UNSUBSCRIBE, Some(params))
            .await?;

        match resp {
            Response::Ok(_) => Ok(()),
            Response::Err(err) => Err(err.error.into()),
        }
    }

    /// Maps the `handler` to a specific `event`
    pub fn subscribe<E, F, R>(&mut self, event: E, handler: F)
    where
        E: Into<String>,
        F: Fn(Notification) -> R + Clone + Send + Sync + 'static,
        R: Future<Output = ()> + Send,
    {
        self.options
            .notification_handler
            .get_or_insert_default()
            .subscribe(event, handler);
    }

    /// Unsubscribe a handler from the `event`
    pub fn unsubscribe(&mut self, event: impl AsRef<str>) {
        if let Some(notification_handler) = &self.options.notification_handler {
            notification_handler.unsubscribe(event);
        }
    }

    /// Returns whether the server is configured to send the "notifications/resources/updated"
    #[inline]
    fn is_resource_subscription_supported(&self) -> bool {
        self.server_capabilities
            .as_ref()
            .and_then(|cap| cap.resources.as_ref())
            .is_some_and(|res| res.subscribe)
    }

    /// Returns whether the server is configured to send the "notifications/resources/list_changed"
    #[inline]
    fn is_resource_list_changed_supported(&self) -> bool {
        self.server_capabilities
            .as_ref()
            .and_then(|cap| cap.resources.as_ref())
            .is_some_and(|res| res.list_changed)
    }

    /// Returns whether the server is configured to send the "notifications/tools/list_changed"
    #[inline]
    fn is_tools_list_changed_supported(&self) -> bool {
        self.server_capabilities
            .as_ref()
            .and_then(|cap| cap.tools.as_ref())
            .is_some_and(|tool| tool.list_changed)
    }

    /// Returns whether the server is configured to send the "notifications/prompts/list_changed"
    #[inline]
    fn is_prompts_list_changed_supported(&self) -> bool {
        self.server_capabilities
            .as_ref()
            .and_then(|cap| cap.prompts.as_ref())
            .is_some_and(|prompt| prompt.list_changed)
    }

    /// Returns whether the client has elicitation capabilities
    #[inline]
    #[cfg(feature = "legacy-spec")]
    fn is_elicitation_supported(&self) -> bool {
        self.options.elicitation_capability.as_ref().is_some()
    }

    /// Returns whether the client has task augmentation capabilities
    #[inline]
    #[cfg(feature = "tasks")]
    fn is_client_supports_tasks(&self) -> bool {
        self.options.tasks_capability.as_ref().is_some()
    }

    /// Resolves the server's tasks capability from the negotiated server
    /// capabilities. Pre-2026-07-28 it is the top-level `tasks` field; under
    /// MCP 2026-07-28 tasks are an extension, so it is read from
    /// `capabilities.extensions["io.modelcontextprotocol/tasks"]`.
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn server_tasks_capability(&self) -> Option<crate::types::ServerTasksCapability> {
        self.server_capabilities
            .as_ref()
            .and_then(|c| c.tasks.clone())
    }

    /// Resolves the server's tasks capability from the negotiated server
    /// capabilities (MCP 2026-07-28 build).
    ///
    /// Tasks are an extension here, so the one place they can be advertised is
    /// `capabilities.extensions["io.modelcontextprotocol/tasks"]`. The
    /// pre-2026-07-28 top-level `tasks` field is deliberately not read: it can
    /// only come from a peer reached through the dual-mode fallback, whose task
    /// protocol this build does not speak (see
    /// [`Self::is_server_supports_tasks`]), so resolving it would only ever
    /// promise support that cannot be delivered.
    #[cfg(all(feature = "tasks", not(feature = "legacy-spec")))]
    fn server_tasks_capability(&self) -> Option<crate::types::ServerTasksCapability> {
        self.server_capabilities
            .as_ref()?
            .extensions
            .as_ref()
            .and_then(|ext| ext.get(crate::types::task::TASKS_EXTENSION_ID))
            .and_then(|v| serde_json::from_value(v.clone()).ok())
    }

    /// Returns whether the server has task augmentation capabilities
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_server_supports_tasks(&self) -> bool {
        self.server_tasks_capability().is_some()
    }

    /// Returns whether the server has task augmentation capabilities
    ///
    /// A peer reached through the dual-mode fallback speaks the *legacy* task
    /// protocol -- different method set (`tasks/result`, `tasks/list`), a
    /// nested `CreateTaskResult`, a differently named status notification --
    /// and none of that wire surface is compiled into this build. It is
    /// reported as unsupported rather than answered with 2026-07-28 messages
    /// it cannot read; talking tasks to a legacy server needs a `legacy-spec`
    /// build, or the peers must simply agree on a generation. The peer check
    /// is belt-and-braces on top of
    /// [`Self::server_tasks_capability`](Self::server_tasks_capability) only
    /// reading the 2026-07-28 form: it also covers a peer that advertises the
    /// extension and then falls back.
    #[inline]
    #[cfg(all(feature = "tasks", not(feature = "legacy-spec")))]
    fn is_server_supports_tasks(&self) -> bool {
        !self.is_legacy_peer() && self.server_tasks_capability().is_some()
    }

    /// Returns whether the client supports cancelling tasks
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_client_support_cancelling_tasks(&self) -> bool {
        self.options
            .tasks_capability
            .as_ref()
            .is_some_and(|c| c.cancel.is_some())
    }

    /// Returns whether the server supports cancelling tasks
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_server_support_cancelling_tasks(&self) -> bool {
        self.server_tasks_capability()
            .is_some_and(|c| c.cancel.is_some())
    }

    /// Returns whether the server supports retrieving a task list
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_server_support_task_list(&self) -> bool {
        self.server_tasks_capability()
            .is_some_and(|c| c.list.is_some())
    }

    /// Returns whether the client supports retrieving a task list
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_client_support_task_list(&self) -> bool {
        self.options
            .tasks_capability
            .as_ref()
            .is_some_and(|c| c.list.is_some())
    }

    /// Returns whether the server supports task-augmented tools
    ///
    /// Under MCP 2026-07-28 the Tasks extension capability carries no
    /// per-request settings: a peer that advertises the extension at all
    /// accepts task-augmented requests, and the server decides per request
    /// whether to defer. A peer that fell back to the legacy protocol is
    /// excluded -- see [`Self::is_server_supports_tasks`].
    #[inline]
    #[cfg(all(feature = "tasks", not(feature = "legacy-spec")))]
    fn is_server_support_call_tool_with_tasks(&self) -> bool {
        self.is_server_supports_tasks()
    }

    /// Returns whether the server supports task-augmented tools
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    fn is_server_support_call_tool_with_tasks(&self) -> bool {
        self.server_tasks_capability()
            .and_then(|c| c.requests)
            .and_then(|r| r.tools)
            .is_some_and(|t| t.call.is_some())
    }

    /// Sends a request to the MCP server
    #[inline]
    async fn send_request(&mut self, req: Request) -> Result<Response, Error> {
        // Checked at the send seam rather than in `call_tool`, so every way of
        // reaching a tool -- the plain call, the task builder -- goes past it.
        #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
        if let Some(err) = self.blocked_tool_error(&req) {
            return Err(err);
        }

        #[cfg(not(feature = "legacy-spec"))]
        {
            // A legacy peer (dual-mode fallback) never speaks MRTR -- its
            // requests take the plain path, elicitation rides the legacy
            // server-push channel instead.
            if self.is_legacy_peer() {
                return self.plain_send_request(req).await;
            }
            self.run_with_mrtr(req).await
        }
        #[cfg(feature = "legacy-spec")]
        {
            self.plain_send_request(req).await
        }
    }

    /// Sends a request without the MRTR loop.
    #[inline]
    async fn plain_send_request(&mut self, req: Request) -> Result<Response, Error> {
        let resp = self
            .handler
            .as_mut()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?
            .send_request(req)
            .await?;
        #[cfg(not(feature = "legacy-spec"))]
        self.record_server_info(&resp);
        Ok(resp)
    }

    /// Picks up `io.modelcontextprotocol/serverInfo` from a result's `_meta`.
    ///
    /// Under MCP 2026-07-28 the server identifies itself on every result rather
    /// than once in a handshake, so the first result that carries it is what
    /// populates [`Self::server_info`].
    #[cfg(not(feature = "legacy-spec"))]
    fn record_server_info(&mut self, resp: &Response) {
        if self.server_info.is_some() {
            return;
        }
        let Response::Ok(ok) = resp else { return };
        if let Some(info) = ok
            .result
            .get("_meta")
            .and_then(|m| m.get("io.modelcontextprotocol/serverInfo"))
            .and_then(|v| serde_json::from_value::<Implementation>(v.clone()).ok())
        {
            self.server_info = Some(info);
        }
    }

    /// Sends a request and transparently drives the MRTR loop: while the
    /// server responds with an `input_required` result, fulfil each
    /// elicitation via the configured handler and re-issue the original
    /// request (new id) with `inputResponses` + the echoed `requestState`.
    #[cfg(not(feature = "legacy-spec"))]
    async fn run_with_mrtr(&mut self, req: Request) -> Result<Response, Error> {
        let max_rounds = self.options.max_mrtr_rounds;
        let method = req.method.clone();
        let original_params = req.params.clone();
        let mrtr_method = shared::is_mrtr_method(method.as_str());

        let mut req = req;
        self.apply_client_meta(&mut req, None, None);

        // The budget counts re-issue rounds, not the initial send, so allow the
        // first attempt plus `max_rounds` retries. `0..=max_rounds` (vs.
        // `max_rounds + 1`) also avoids overflow at `usize::MAX`.
        for _ in 0..=max_rounds {
            let resp = self
                .handler
                .as_mut()
                .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?
                .send_request(req)
                .await?;
            self.record_server_info(&resp);

            // MRTR only applies to success results carrying the
            // `input_required` discriminator; anything else -- including a
            // result with no `resultType` at all -- is final.
            let input_required_result = match &resp {
                Response::Ok(ok)
                    if mrtr_method
                        && resp.result_type() == Some(crate::types::ResultType::InputRequired) =>
                {
                    serde_json::from_value::<crate::types::mrtr::InputRequiredResult>(
                        ok.result.clone(),
                    )
                    .map_err(Error::from)?
                }
                _ => return Ok(resp),
            };
            let ir = input_required_result;

            let mut input_responses = crate::types::mrtr::InputResponses::new();
            if let Some(reqs) = ir.input_requests {
                for (key, request) in reqs {
                    input_responses.insert(key, self.fulfil_input(request).await?);
                }
            }

            let new_id = self.generate_id()?;
            let mut retry = Request::new(Some(new_id), method.clone(), original_params.clone());
            self.apply_client_meta(&mut retry, Some(input_responses), ir.request_state);
            req = retry;
        }

        Err(Error::new(
            ErrorCode::InternalError,
            "MRTR exceeded the maximum number of rounds",
        ))
    }

    /// Sets `clientInfo` + MRTR capability `_meta` on a request, and optionally
    /// the MRTR `inputResponses` / `requestState`, preserving existing `_meta`.
    ///
    /// Also populates W3C Trace Context (`traceparent` / `tracestate`) from the
    /// configured [`trace_context_provider`](crate::client::options::McpOptions::with_trace_context_provider),
    /// when installed. This is the single assembly point for outbound 2026-07-28 `_meta`,
    /// so both single sends (via [`Self::run_with_mrtr`]) and batched requests
    /// (via [`Self::run_batch_with_mrtr`]) carry trace context.
    #[cfg(not(feature = "legacy-spec"))]
    fn apply_client_meta(
        &self,
        req: &mut Request,
        input_responses: Option<crate::types::mrtr::InputResponses>,
        request_state: Option<String>,
    ) {
        let mut meta = req.meta().unwrap_or_default();
        meta.client_info = Some(self.options.implementation.clone());
        // Required on every request under MCP 2026-07-28, and it must agree
        // with the `MCP-Protocol-Version` header the HTTP transport sets.
        meta.protocol_version = Some(self.expected_protocol_ver().to_string());
        // Each flag reflects what this client can actually fulfil right now: a
        // configured handler for elicitation/sampling, and -- since roots are
        // data rather than a handler -- a declared roots capability. That is
        // either an explicit `with_roots(..)` or simply having roots, and it
        // deliberately stays true for an empty list: an empty
        // `ListRootsResult` is a valid answer, so a client that opted in must
        // not be gated out of being asked.
        meta.client_capabilities = Some(crate::types::mrtr::ClientMrtrCapabilities {
            elicitation: self.options.elicitation_handler.is_some(),
            sampling: self.options.sampling_handler.is_some(),
            roots: self.options.roots_capability().is_some(),
        });

        if input_responses.is_some() {
            meta.input_responses = input_responses;
        }

        if request_state.is_some() {
            meta.request_state = request_state;
        }

        if let Some(provider) = self.options.trace_context_provider.as_ref()
            && let Some(tc) = provider()
        {
            meta.traceparent = Some(tc.traceparent);
            meta.tracestate = tc.tracestate;
            meta.baggage = tc.baggage;
        }

        // Request-scoped logging level (replaces the removed `logging/setLevel`).
        if self.options.log_level.is_some() {
            meta.log_level = self.options.log_level;
        }

        req.set_meta(meta);
    }

    /// Applies the initial per-request 2026-07-28 client metadata (`clientInfo` /
    /// `clientCapabilities`, plus trace context) to every [`Request`] in a
    /// batch. The MRTR re-run fields (`inputResponses` / `requestState`) stay
    /// `None` here -- they are filled per request on each retry round by
    /// [`Self::run_batch_with_mrtr`]. Notifications are left untouched.
    #[cfg(not(feature = "legacy-spec"))]
    fn apply_client_meta_to_batch(&self, items: &mut [MessageEnvelope]) {
        for envelope in items {
            if let MessageEnvelope::Request(req) = envelope {
                self.apply_client_meta(req, None, None);
            }
        }
    }

    /// Fulfils one server-requested input, whatever its kind, and returns the
    /// raw result to echo back under the request's key.
    ///
    /// Sampling and roots are fulfilled here, on the MRTR loop -- *not* as
    /// server-initiated pushes: under MCP 2026-07-28 there is no such channel. The
    /// client only ever gets asked for a kind it declared in
    /// [`ClientMrtrCapabilities`](crate::types::mrtr::ClientMrtrCapabilities),
    /// so a missing handler here means the server ignored those flags.
    #[cfg(not(feature = "legacy-spec"))]
    async fn fulfil_input(
        &self,
        request: crate::types::mrtr::InputRequest,
    ) -> Result<serde_json::Value, Error> {
        use crate::types::mrtr::InputRequest;

        #[allow(deprecated)]
        let value = match request {
            InputRequest::Elicitation(params) => match self.options.elicitation_handler.clone() {
                Some(handler) => serde_json::to_value(handler(params).await)?,
                None => return Err(no_fulfiller("elicitation")),
            },
            InputRequest::Sampling(params) => match self.options.sampling_handler.clone() {
                Some(handler) => serde_json::to_value(handler(*params).await)?,
                None => return Err(no_fulfiller("sampling")),
            },
            // Roots are configured data, not a handler: the client answers
            // from the list it was built with.
            InputRequest::Roots(_) => serde_json::to_value(crate::types::root::ListRootsResult {
                roots: self.options.roots(),
                meta: None,
            })?,
        };
        Ok(value)
    }

    /// Creates a [`BatchBuilder`] for sending multiple requests in a single batch.
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///     client.connect().await?;
    ///
    ///     let responses = client
    ///         .batch()
    ///         .list_tools()
    ///         .list_prompts()
    ///         .send()
    ///         .await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    pub fn batch(&mut self) -> BatchBuilder<'_> {
        BatchBuilder {
            client: self,
            items: Vec::new(),
        }
    }

    /// Returns a [`TaskBuilder`] for constructing a task-augmented request.
    ///
    /// Chain setters such as [`TaskBuilder::with_ttl`] to configure the task,
    /// then call [`TaskBuilder::call_tool`] to execute.
    ///
    /// # Example
    /// ```no_run
    /// use neva::client::Client;
    /// use neva::error::Error;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Error> {
    ///     let mut client = Client::new();
    ///     client.connect().await?;
    ///
    ///     let result = client
    ///         .task()
    ///         .with_ttl(5000)
    ///         .call_tool("echo", [("message", "Hello MCP!")])
    ///         .await?;
    ///
    ///     client.disconnect().await
    /// }
    /// ```
    #[cfg(feature = "tasks")]
    pub fn task(&mut self) -> TaskBuilder<'_> {
        TaskBuilder {
            client: self,
            metadata: TaskMetadata::default(),
        }
    }

    /// Sends a batch of messages to the MCP server and awaits all responses.
    ///
    /// Items that are [`MessageEnvelope::Request`] each get a response slot in
    /// the returned `Vec`, in the same order they appear in `items`.
    /// [`MessageEnvelope::Notification`] items are sent fire-and-forget and
    /// produce no slot.
    ///
    /// All in-flight requests are awaited concurrently; a failure in one
    /// does not cancel the others.
    ///
    /// # Errors
    /// Returns [`Error`] if the client is not connected, the batch is empty,
    /// or any response channel is closed or times out.
    pub async fn call_batch(
        &mut self,
        items: Vec<MessageEnvelope>,
    ) -> Result<Vec<Response>, Error> {
        // One blocked tool fails the whole batch, the same as a duplicate id
        // does: the batch is one write, and there is no way to drop a single
        // entry from it without silently changing what the caller asked for.
        #[cfg(all(feature = "http-client", not(feature = "legacy-spec")))]
        if let Some(err) = items.iter().find_map(|env| match env {
            MessageEnvelope::Request(req) => self.blocked_tool_error(req),
            _ => None,
        }) {
            return Err(err);
        }

        // Under MCP 2026-07-28 a batched request may elicit just like a single send, so
        // the batch is driven through the same MRTR retry loop (see
        // `run_batch_with_mrtr`) rather than returning the protocol-intermediate
        // `input_required` result as final. A legacy peer (dual-mode
        // fallback) never speaks MRTR and takes the plain path.
        #[cfg(not(feature = "legacy-spec"))]
        {
            if !self.is_legacy_peer() {
                return self.run_batch_with_mrtr(items).await;
            }
        }
        let handler = self
            .handler
            .as_mut()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?;

        let request_timeout = handler.timeout();
        let pending = handler.pending().clone();
        let token = handler.cancellation();
        let receivers = handler.send_batch(items).await?;

        collect_batch_responses(receivers, &pending, request_timeout, token)
            .await
            .into_iter()
            .collect()
    }

    /// Drives the MRTR retry loop across an entire batch.
    ///
    /// Each batched [`Request`] that elicits -- the server replies with an
    /// `input_required` result -- is fulfilled via the configured elicitation
    /// handler and re-issued (carrying `inputResponses` + the echoed
    /// `requestState`) alongside any other still-eliciting requests, so the
    /// whole batch is driven to completion in lock-step rounds. One transport
    /// write per round preserves the batching benefit; final
    /// (non-`input_required`) responses are retained and not re-sent. Each
    /// request keeps its slot in the returned `Vec`, in input order;
    /// notifications (and any non-request envelopes) are sent once, in the
    /// first round, and produce no slot.
    #[cfg(not(feature = "legacy-spec"))]
    async fn run_batch_with_mrtr(
        &mut self,
        items: Vec<MessageEnvelope>,
    ) -> Result<Vec<Response>, Error> {
        let max_rounds = self.options.max_mrtr_rounds;

        // A per-request slot: either still eliciting (carrying what is needed to
        // re-issue) or resolved to its final response.
        enum Slot {
            Pending {
                method: String,
                original_params: Option<serde_json::Value>,
                req: Option<Request>,
            },
            Done(Response),
        }

        // Seed round-0 metadata (`clientInfo` / `clientCapabilities` / trace
        // context) on every request via the shared assembly path, then split
        // requests (which get ordered slots) from fire-and-forget extras
        // (notifications) sent only in the first round.
        let mut items = items;
        self.apply_client_meta_to_batch(&mut items);

        let mut slots: Vec<Slot> = Vec::new();
        let mut extras: Vec<MessageEnvelope> = Vec::new();
        for envelope in items {
            match envelope {
                MessageEnvelope::Request(req) => slots.push(Slot::Pending {
                    method: req.method.clone(),
                    original_params: req.params.clone(),
                    req: Some(req),
                }),
                other => extras.push(other),
            }
        }

        // No requests to drive through MRTR: send any extras (notifications)
        // once and let `send_batch` surface the same connection-closed /
        // empty-batch errors a non-eliciting batch would.
        if slots.is_empty() {
            let handler = self
                .handler
                .as_mut()
                .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?;
            let request_timeout = handler.timeout();
            let pending = handler.pending().clone();
            let token = handler.cancellation();
            let receivers = handler.send_batch(extras).await?;
            return collect_batch_responses(receivers, &pending, request_timeout, token)
                .await
                .into_iter()
                .collect();
        }

        // Round 0 is the initial batch send; rounds `1..=max_rounds` are the
        // re-issues the budget allows (the cap counts retries, not the first
        // send). `0..=max_rounds` also avoids overflow at `usize::MAX`.
        for round in 0..=max_rounds {
            // Collect this round's outgoing requests; notifications ride along once.
            let mut envelopes: Vec<MessageEnvelope> = Vec::new();
            if round == 0 {
                envelopes.append(&mut extras);
            }

            let mut round_slots: Vec<usize> = Vec::new();
            for (i, slot) in slots.iter_mut().enumerate() {
                if let Slot::Pending { req, .. } = slot
                    && let Some(request) = req.take()
                {
                    round_slots.push(i);
                    envelopes.push(MessageEnvelope::Request(request));
                }
            }

            if round_slots.is_empty() {
                break;
            }

            // One transport write; await this round's replies concurrently.
            let handler = self
                .handler
                .as_mut()
                .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?;

            let request_timeout = handler.timeout();
            let pending = handler.pending().clone();
            let token = handler.cancellation();
            let receivers = handler.send_batch(envelopes).await?;
            let responses =
                collect_batch_responses(receivers, &pending, request_timeout, token).await;

            // `responses` aligns with `round_slots`: `send_batch` preserves
            // request order and extras produce no receiver. Final responses fill
            // their slot; `input_required` ones are fulfilled and re-issued.
            for (slot_i, resp) in round_slots.into_iter().zip(responses) {
                let resp = resp?;
                self.record_server_info(&resp);
                let (method, original_params) = match &slots[slot_i] {
                    Slot::Pending {
                        method,
                        original_params,
                        ..
                    } => (method.clone(), original_params.clone()),
                    Slot::Done(_) => unreachable!("a round slot is always pending"),
                };

                let is_input_required = shared::is_mrtr_method(method.as_str())
                    && resp.result_type() == Some(crate::types::ResultType::InputRequired);

                if !is_input_required {
                    slots[slot_i] = Slot::Done(resp);
                    continue;
                }

                let ir = match &resp {
                    Response::Ok(ok) => serde_json::from_value::<
                        crate::types::mrtr::InputRequiredResult,
                    >(ok.result.clone())
                    .map_err(Error::from)?,
                    Response::Err(_) => unreachable!("input_required is a success result"),
                };

                let mut input_responses = crate::types::mrtr::InputResponses::new();
                if let Some(reqs) = ir.input_requests {
                    for (key, request) in reqs {
                        input_responses.insert(key, self.fulfil_input(request).await?);
                    }
                }

                let new_id = self.generate_id()?;
                let mut retry = Request::new(Some(new_id), method, original_params);

                self.apply_client_meta(&mut retry, Some(input_responses), ir.request_state);

                if let Slot::Pending { req, .. } = &mut slots[slot_i] {
                    *req = Some(retry);
                }
            }
        }

        // Assemble in slot order; a slot still pending exhausted the rounds.
        let mut out = Vec::with_capacity(slots.len());
        for slot in slots {
            match slot {
                Slot::Done(resp) => out.push(resp),
                Slot::Pending { .. } => {
                    return Err(Error::new(
                        ErrorCode::InternalError,
                        "MRTR exceeded the maximum number of rounds",
                    ));
                }
            }
        }
        Ok(out)
    }

    /// Sends a response to the MCP server
    ///
    /// Only the legacy profile has server->client requests to answer.
    #[inline]
    #[cfg(all(feature = "tasks", feature = "legacy-spec"))]
    async fn send_response(&mut self, req: Response) -> Result<(), Error> {
        self.handler
            .as_mut()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?
            .send_response(req)
            .await;
        Ok(())
    }

    /// Sends a notification to the MCP server
    #[inline]
    async fn send_notification(
        &mut self,
        method: &str,
        params: Option<serde_json::Value>,
    ) -> Result<(), Error> {
        let notification = Notification::new(method, params);
        self.handler
            .as_mut()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))?
            .send_notification(notification)
            .await
    }

    #[cfg(feature = "tracing")]
    fn register_tracing_notification_handlers(&mut self) {
        use crate::types::notification::commands::*;

        self.subscribe(MESSAGE, Self::default_notification_handler);
        self.subscribe(STDERR, Self::default_notification_handler);
        self.subscribe(PROGRESS, Self::default_notification_handler);
    }

    #[cfg(feature = "tracing")]
    async fn default_notification_handler(notification: Notification) {
        notification.write();
    }

    /// Generates a new [`RequestId`]
    #[inline]
    fn generate_id(&self) -> Result<RequestId, Error> {
        self.handler
            .as_ref()
            .ok_or_else(|| Error::new(ErrorCode::InternalError, "Connection closed"))
            .map(|h| h.next_id())
    }

    /// Cancels the transport and clears connection state without sending a
    /// notification. Used when initialization fails after the transport has
    /// already been started (e.g. protocol version mismatch in `init()`).
    #[inline]
    fn cancel_transport(&mut self) {
        if let Some(token) = self.cancellation_token.take() {
            token.cancel();
        }
        self.handler = None;
    }

    #[inline]
    fn wait_for_shutdown_signal(&mut self) {
        if let Some(token) = self.cancellation_token.clone() {
            shared::wait_for_shutdown_signal(token);
        };
    }

    #[cfg(feature = "tasks")]
    pub(crate) fn ensure_tasks_supported(&self) {
        assert!(
            self.is_client_supports_tasks(),
            "Client does not support task-augmented requests. You may configure it with `Client::with_options(|opt| opt.with_tasks(...))` method."
        );

        assert!(
            self.is_server_supports_tasks(),
            "Server does not support task-augmented requests."
        );
    }
}

#[cfg(all(feature = "tasks", not(feature = "legacy-spec")))]
impl shared::TaskApi for Client {
    /// Retrieves the full task state: status plus, depending on it, the
    /// outstanding input requests, the terminal result, or the error.
    async fn get_task(&mut self, id: impl Into<String>) -> Result<DetailedTask, Error> {
        let params = GetTaskRequestParams { id: id.into() };
        self.command(crate::types::task::commands::GET, Some(params))
            .await?
            .into_result()
    }

    /// Submits responses to a task's outstanding input requests.
    async fn update_task(
        &mut self,
        id: impl Into<String>,
        responses: crate::types::mrtr::InputResponses,
    ) -> Result<(), Error> {
        let params = UpdateTaskRequestParams {
            id: id.into(),
            input_responses: responses,
        };
        self.command(crate::types::task::commands::UPDATE, Some(params))
            .await
            .map(|_| ())
    }

    /// Cancels a task that is currently running.
    ///
    /// The reply is an empty acknowledgement: cancellation is cooperative, so
    /// the task may still reach a non-`cancelled` terminal status. Poll
    /// `get_task` to learn the outcome.
    async fn cancel_task(&mut self, id: impl Into<String>) -> Result<(), Error> {
        let params = CancelTaskRequestParams { id: id.into() };
        self.command(crate::types::task::commands::CANCEL, Some(params))
            .await
            .map(|_| ())
    }

    /// Answers one outstanding input request with the client's configured
    /// handler for that kind.
    async fn fulfil_input(
        &mut self,
        request: &crate::types::mrtr::InputRequest,
    ) -> Result<serde_json::Value, Error> {
        use crate::types::mrtr::InputRequest;

        match request {
            InputRequest::Elicitation(params) => {
                let handler = self.options.elicitation_handler.as_ref().ok_or_else(|| {
                    Error::new(
                        ErrorCode::InvalidRequest,
                        "Client has no elicitation handler. Configure one with `Client::map_elicitation(...)`.",
                    )
                })?;
                let result = handler(params.clone()).await;
                serde_json::to_value(result).map_err(Into::into)
            }
            other => Err(Error::new(
                ErrorCode::InvalidRequest,
                format!(
                    "Client cannot fulfil `{}` input requests on the task substrate",
                    other.method()
                ),
            )),
        }
    }
}

#[cfg(all(feature = "tasks", feature = "legacy-spec"))]
impl shared::TaskApi for Client {
    /// Retrieves task result. If the task is not completed yet, waits until it completes or cancels.
    async fn get_task_result<T>(&mut self, id: impl Into<String>) -> Result<T, Error>
    where
        T: DeserializeOwned,
    {
        let params = GetTaskPayloadRequestParams { id: id.into() };
        self.command(crate::types::task::commands::RESULT, Some(params))
            .await?
            .into_result()
    }

    /// Retrieve task status
    async fn get_task(&mut self, id: impl Into<String>) -> Result<Task, Error> {
        let params = GetTaskRequestParams { id: id.into() };
        self.command(crate::types::task::commands::GET, Some(params))
            .await?
            .into_result()
    }

    /// Cancels a task that is currently running
    ///
    /// # Panics
    /// If the client or server does not support cancelling tasks
    async fn cancel_task(&mut self, id: impl Into<String>) -> Result<Task, Error> {
        assert!(
            self.is_client_support_cancelling_tasks(),
            "Client does not support cancelling tasks.  You may configure it with `Client::with_options(|opt| opt.with_tasks(...))` method."
        );

        assert!(
            self.is_server_support_cancelling_tasks(),
            "Server does not support cancelling tasks."
        );

        let params = CancelTaskRequestParams { id: id.into() };
        self.command(crate::types::task::commands::CANCEL, Some(params))
            .await?
            .into_result()
    }

    /// Retrieves a list of tasks
    ///
    /// # Panics
    /// If the client or server does not support retrieving a task list
    async fn list_tasks(&mut self, cursor: Option<Cursor>) -> Result<ListTasksResult, Error> {
        assert!(
            self.is_client_support_task_list(),
            "Client does not support retrieving a task list.  You may configure it with `Client::with_options(|opt| opt.with_tasks(...))` method."
        );

        assert!(
            self.is_server_support_task_list(),
            "Server does not support retrieving a task list."
        );

        let params = ListTasksRequestParams { cursor };
        self.command(crate::types::task::commands::LIST, Some(params))
            .await?
            .into_result()
    }

    async fn handle_input(&mut self, id: &str, params: TaskPayload) -> Result<(), Error> {
        let params = params.to::<ElicitRequestParams>()?;
        if let Some(handler) = &self.options.elicitation_handler {
            use crate::types::IntoResponse;

            let result = handler(params).await.with_related_task(id);

            let id = id.parse::<RequestId>().expect("Invalid Request Id");

            self.send_response(result.into_response(id)).await?;
        }
        Ok(())
    }
}

/// Awaits a batch's per-request receivers concurrently, returning one result
/// per receiver in input order, with the same per-request timeout and pending
/// cleanup as a single [`RequestHandler::send_request`].
///
/// Uses `join_all` (not `try_join_all`) so every future runs to completion: the
/// timeout-cleanup branch (`pending.pop`) executes for each timed-out request
/// even when another request in the same batch has already failed.
async fn collect_batch_responses(
    receivers: Vec<(
        RequestId,
        tokio::sync::oneshot::Receiver<crate::shared::PendingResponse>,
    )>,
    pending: &crate::shared::RequestQueue,
    request_timeout: std::time::Duration,
    token: tokio_util::sync::CancellationToken,
) -> Vec<Result<Response, Error>> {
    use futures_util::future::join_all;

    let futures = receivers.into_iter().map(|(id, rx)| {
        let pending = pending.clone();
        let token = token.clone();
        async move {
            tokio::select! {
                biased;
                // The transport died (or a shutdown signal cancelled it)
                // -- no response is coming for any receiver.
                _ = token.cancelled() => {
                    let _ = pending.pop(&id);
                    Err(Error::new(ErrorCode::InternalError, "Connection closed"))
                }
                result = tokio::time::timeout(request_timeout, rx) => match result {
                    Ok(Ok(crate::shared::PendingResponse::Response(resp))) => Ok(resp),
                    Ok(Ok(crate::shared::PendingResponse::Timeout)) => {
                        Err(Error::new(ErrorCode::Timeout, "Batch request timed out"))
                    }
                    Ok(Err(_)) => Err(Error::new(
                        ErrorCode::InternalError,
                        "Response channel closed",
                    )),
                    Err(_) => {
                        let _ = pending.pop(&id);
                        Err(Error::new(ErrorCode::Timeout, "Batch request timed out"))
                    }
                }
            }
        }
    });

    join_all(futures).await
}

/// The error for an input kind the server asked for but this client has no
/// fulfiller for -- only reachable if the server ignored the declared
/// [`ClientMrtrCapabilities`](crate::types::mrtr::ClientMrtrCapabilities).
#[cfg(not(feature = "legacy-spec"))]
fn no_fulfiller(kind: &str) -> Error {
    Error::new(
        ErrorCode::InvalidRequest,
        format!("server requested {kind} but no handler is configured"),
    )
}

/// Whether a `server/discover` failure means "this server doesn't speak
/// the 2026-07-28 protocol" -- the dual-mode fallback triggers only then.
///
/// * `MethodNotFound` -- a legacy server rejecting the unknown method
///   (neva's own legacy build answers exactly this);
/// * `InvalidRequest` -- strict servers rejecting the 2026-07-28 request shape;
/// * `ParseError` -- a non-JSON-RPC reply (an HTTP 4xx page) or an error
///   code outside neva's `ErrorCode` set (e.g. the TS SDK's `-32000`
///   "server not initialized"), both of which surface as parse failures.
///
/// Network-level failures (`Timeout`, `InternalError`/"Connection
/// closed") are *not* triggers: the server never answered, so falling
/// back would only mask the outage.
#[cfg(not(feature = "legacy-spec"))]
fn is_fallback_trigger(err: &Error) -> bool {
    matches!(
        err.code,
        ErrorCode::MethodNotFound | ErrorCode::InvalidRequest | ErrorCode::ParseError
    )
}

#[inline]
fn make_handler<F, R, P, O>(handler: F) -> Handler<P, O>
where
    F: Fn(P) -> R + Clone + Send + Sync + 'static,
    R: Future + Send,
    R::Output: Into<O>,
    P: Send + 'static,
    O: Send + 'static,
{
    Arc::new(move |params: P| {
        let handler = handler.clone();
        Box::pin(async move { handler(params).await.into() })
    })
}

type Handler<P, O> =
    Arc<dyn Fn(P) -> std::pin::Pin<Box<dyn Future<Output = O> + Send>> + Send + Sync>;

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn call_batch_requires_connected_client() {
        let mut client = Client::new();
        let result = client.call_batch(vec![]).await;
        assert!(
            result.is_err(),
            "disconnected client should return an error"
        );
    }

    #[cfg(not(feature = "legacy-spec"))]
    #[test]
    fn batch_injects_rc_client_meta_per_request() {
        use serde_json::json;

        let mut client = Client::new();
        // Registering an elicitation handler makes the client declare
        // `clientCapabilities.elicitation = true`.
        client.map_elicitation(|_params: ElicitRequestParams| async { ElicitResult::accept() });

        let req = Request::new(
            Some(RequestId::Number(1)),
            "tools/call",
            Some(json!({ "name": "greet", "arguments": {} })),
        );
        let mut items = vec![
            MessageEnvelope::Request(req),
            // Notifications must be left untouched.
            MessageEnvelope::Notification(Notification::new("notifications/progress", None)),
        ];

        client.apply_client_meta_to_batch(&mut items);

        let MessageEnvelope::Request(req) = &items[0] else {
            panic!("first item must be a request");
        };
        let meta = &req.params.as_ref().expect("params present")["_meta"];
        // Without this injection a batched eliciting tools/call is rejected as
        // if the client did not support elicitation.
        assert_eq!(
            meta["io.modelcontextprotocol/clientCapabilities"]["elicitation"],
            json!(true)
        );
        assert!(meta["io.modelcontextprotocol/clientInfo"].is_object());
        assert_eq!(
            meta["io.modelcontextprotocol/protocolVersion"],
            json!("2026-07-28")
        );

        // The notification carries no params/_meta.
        let MessageEnvelope::Notification(notif) = &items[1] else {
            panic!("second item must be a notification");
        };
        assert!(notif.params.is_none());
    }

    /// A configured trace-context provider is invoked during 2026-07-28 metadata
    /// assembly, so `_meta.traceparent`/`tracestate` reach the wire alongside
    /// `clientInfo` -- for both single sends and (via the same path) batches.
    #[cfg(not(feature = "legacy-spec"))]
    #[test]
    fn apply_client_meta_injects_trace_context() {
        use crate::client::options::TraceContext;
        use serde_json::json;

        let client = Client::new().with_options(|o| {
            o.with_trace_context_provider(|| {
                Some(TraceContext {
                    traceparent: "tp".into(),
                    tracestate: Some("ts".into()),
                    baggage: Some("bg".into()),
                })
            })
        });

        let mut req = Request::new(
            Some(RequestId::Number(1)),
            "tools/call",
            Some(json!({ "name": "greet", "arguments": {} })),
        );
        client.apply_client_meta(&mut req, None, None);

        let meta = &req.params.as_ref().expect("params present")["_meta"];
        assert_eq!(meta["traceparent"], json!("tp"));
        assert_eq!(meta["tracestate"], json!("ts"));
        // Trace context is assembled alongside the rest of the 2026-07-28 metadata.
        assert!(meta["io.modelcontextprotocol/clientInfo"].is_object());
    }

    /// With no provider installed, no trace fields are emitted.
    #[cfg(not(feature = "legacy-spec"))]
    #[test]
    fn apply_client_meta_omits_trace_context_without_provider() {
        use serde_json::json;

        let client = Client::new();
        let mut req = Request::new(
            Some(RequestId::Number(1)),
            "tools/call",
            Some(json!({ "name": "greet", "arguments": {} })),
        );
        client.apply_client_meta(&mut req, None, None);

        let meta = &req.params.as_ref().expect("params present")["_meta"];
        assert!(meta.get("traceparent").is_none());
        assert!(meta.get("tracestate").is_none());
    }
}

#[cfg(all(test, not(feature = "legacy-spec")))]
mod fallback_trigger_tests {
    use super::*;

    fn err(code: ErrorCode) -> Error {
        Error::new(code, "test")
    }

    #[test]
    fn protocol_level_rejections_trigger_the_fallback() {
        assert!(is_fallback_trigger(&err(ErrorCode::MethodNotFound)));
        assert!(is_fallback_trigger(&err(ErrorCode::InvalidRequest)));
        // Non-JSON-RPC replies (HTTP 4xx pages) and unknown error codes
        // (e.g. the TS SDK's -32000) surface as parse failures.
        assert!(is_fallback_trigger(&err(ErrorCode::ParseError)));
    }

    #[test]
    fn transport_failures_do_not_trigger_the_fallback() {
        assert!(!is_fallback_trigger(&err(ErrorCode::Timeout)));
        assert!(!is_fallback_trigger(&err(ErrorCode::InternalError)));
        assert!(!is_fallback_trigger(&err(ErrorCode::InvalidParams)));
    }
}

/// The dual-mode "Done when" (issue #84): a 2026-07-28 client completes calls
/// against a 2025-11-25 server via the `initialize` fallback. The legacy
/// server is a raw-HTTP mock because a legacy neva server cannot exist
/// in an 2026-07-28 build.
#[cfg(all(test, feature = "http-client", not(feature = "legacy-spec")))]
mod dual_mode_tests {
    use super::*;
    use std::sync::Mutex;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::{TcpListener, TcpStream};

    const LEGACY_SESSION_ID: &str = "6f2f0dc8-6a5e-4f6e-9c1a-2b7f9d3f1c11";

    /// Reads one HTTP/1.1 request; returns `(head, body)`.
    async fn read_request(stream: &mut TcpStream) -> Option<(String, String)> {
        let mut buf = Vec::new();
        let mut tmp = [0u8; 2048];
        let header_end = loop {
            let n = stream.read(&mut tmp).await.ok()?;
            if n == 0 {
                return None;
            }
            buf.extend_from_slice(&tmp[..n]);
            if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                break pos + 4;
            }
            if buf.len() > 65536 {
                return None;
            }
        };
        let head = String::from_utf8_lossy(&buf[..header_end]).to_string();
        let content_length = head
            .lines()
            .find_map(|l| {
                l.to_ascii_lowercase()
                    .strip_prefix("content-length:")
                    .map(|v| v.trim().parse::<usize>().ok())
            })
            .flatten()
            .unwrap_or(0);
        while buf.len() < header_end + content_length {
            let n = stream.read(&mut tmp).await.ok()?;
            if n == 0 {
                return None;
            }
            buf.extend_from_slice(&tmp[..n]);
        }
        let body =
            String::from_utf8_lossy(&buf[header_end..header_end + content_length]).to_string();
        Some((head, body))
    }

    async fn write_response(stream: &mut TcpStream, status: &str, extra_headers: &str, body: &str) {
        let resp = format!(
            "HTTP/1.1 {status}\r\n{extra_headers}Content-Length: {}\r\nConnection: keep-alive\r\n\r\n{body}",
            body.len()
        );
        let _ = stream.write_all(resp.as_bytes()).await;
    }

    fn rpc_result(id: &serde_json::Value, result: serde_json::Value) -> String {
        serde_json::json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string()
    }

    /// A minimal 2025-11-25 server: rejects `server/discover` with
    /// `MethodNotFound`, answers `initialize` with a session id, serves
    /// an SSE GET stream and an empty `tools/list`.
    /// How the mock answers `server/discover`: legacy servers reject it
    /// with a JSON-RPC `MethodNotFound` (neva's own legacy build) or a
    /// plain non-JSON-RPC 4xx page (framework routers, the TS SDK's
    /// "server not initialized" family); a future server answers
    /// *successfully* but with a `protocolVersion` this build does not
    /// support -- which must NOT trigger the fallback.
    /// A protected endpoint answering `401` with a non-JSON body must not
    /// trigger the fallback either -- that is an authentication failure,
    /// not evidence of a legacy peer.
    #[derive(Clone, Copy)]
    enum DiscoverReply {
        MethodNotFound,
        Html400,
        UnsupportedVersion,
        Unauthorized401,
        Unavailable503,
    }

    async fn serve_legacy(
        listener: TcpListener,
        log: Arc<Mutex<Vec<String>>>,
        reply: DiscoverReply,
    ) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            let log = log.clone();
            tokio::spawn(async move {
                loop {
                    let Some((head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    log.lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .push(format!("{head}{body}"));

                    if head.starts_with("GET") {
                        let _ = stream
                            .write_all(
                                b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n: hi\n\n",
                            )
                            .await;
                        // Hold the stream open like a real legacy server.
                        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
                        return;
                    }

                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    match msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default()
                    {
                        crate::commands::DISCOVER => match reply {
                            DiscoverReply::MethodNotFound => {
                                let body = serde_json::json!({
                                    "jsonrpc": "2.0",
                                    "id": id,
                                    "error": { "code": -32601, "message": "Method not found" }
                                })
                                .to_string();
                                write_response(
                                    &mut stream,
                                    "200 OK",
                                    "Content-Type: application/json\r\n",
                                    &body,
                                )
                                .await;
                            }
                            DiscoverReply::Html400 => {
                                write_response(
                                    &mut stream,
                                    "400 Bad Request",
                                    "Content-Type: text/html\r\n",
                                    "<html><body>Bad Request</body></html>",
                                )
                                .await;
                            }
                            DiscoverReply::Unavailable503 => {
                                write_response(
                                    &mut stream,
                                    "503 Service Unavailable",
                                    "Content-Type: text/html\r\n",
                                    "<html><body>upstream down</body></html>",
                                )
                                .await;
                            }
                            DiscoverReply::Unauthorized401 => {
                                write_response(
                                    &mut stream,
                                    "401 Unauthorized",
                                    "Content-Type: text/html\r\nWWW-Authenticate: Bearer\r\n",
                                    "<html><body>Unauthorized</body></html>",
                                )
                                .await;
                            }
                            DiscoverReply::UnsupportedVersion => {
                                let body = rpc_result(
                                    &id,
                                    serde_json::json!({
                                        "supportedVersions": ["2099-01-01"],
                                        "capabilities": { "tools": {} }
                                    }),
                                );
                                write_response(
                                    &mut stream,
                                    "200 OK",
                                    "Content-Type: application/json\r\n",
                                    &body,
                                )
                                .await;
                            }
                        },
                        crate::commands::INIT => {
                            let body = rpc_result(
                                &id,
                                serde_json::json!({
                                    "protocolVersion": "2025-11-25",
                                    "capabilities": { "tools": {} },
                                    "serverInfo": { "name": "legacy-mock", "version": "1.0.0" }
                                }),
                            );
                            let headers = format!(
                                "Content-Type: application/json\r\nMcp-Session-Id: {LEGACY_SESSION_ID}\r\n"
                            );
                            write_response(&mut stream, "200 OK", &headers, &body).await;
                        }
                        crate::types::tool::commands::LIST => {
                            let body = rpc_result(&id, serde_json::json!({ "tools": [] }));
                            write_response(
                                &mut stream,
                                "200 OK",
                                "Content-Type: application/json\r\n",
                                &body,
                            )
                            .await;
                        }
                        // notifications (initialized / cancelled)
                        _ => write_response(&mut stream, "202 Accepted", "", "").await,
                    }
                }
            });
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn client_falls_back_to_initialize_against_legacy_server() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(
            listener,
            log.clone(),
            DiscoverReply::MethodNotFound,
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        client
            .connect()
            .await
            .expect("fallback connect must succeed");
        assert!(client.is_legacy_peer(), "peer must be marked legacy");
        assert_eq!(
            client.server_info.as_ref().map(|i| i.name.as_str()),
            Some("legacy-mock")
        );

        let tools = client
            .list_tools(None)
            .await
            .expect("tools/list must work after the fallback");
        assert!(tools.tools.is_empty());

        let log = log
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let discover = log
            .iter()
            .find(|r| r.contains("server/discover"))
            .expect("discover must be attempted first");
        assert!(
            discover
                .to_ascii_lowercase()
                .contains("mcp-protocol-version"),
            "the 2026-07-28 attempt carries the protocol-version header"
        );

        let init = log
            .iter()
            .find(|r| r.contains("\"method\":\"initialize\""))
            .expect("initialize must be sent after the rejection");
        assert!(
            init.contains("2025-11-25"),
            "the fallback negotiates the newest legacy version"
        );

        let list = log
            .iter()
            .find(|r| r.contains(crate::types::tool::commands::LIST))
            .expect("tools/list recorded");
        let list_lower = list.to_ascii_lowercase();
        assert!(
            !list_lower.contains("mcp-protocol-version"),
            "legacy peers must not receive the 2026-07-28 protocol-version header"
        );
        assert!(
            !list_lower.contains("mcp-method:"),
            "legacy peers must not receive 2026-07-28 routing headers"
        );
        assert!(
            list_lower.contains(&format!("mcp-session-id: {LEGACY_SESSION_ID}")),
            "the captured session id must ride every request"
        );
    }

    /// A legacy server that rejects `server/discover` with a plain
    /// non-JSON-RPC 4xx page (no JSON-RPC error at all) must also
    /// trigger the fallback: the transport completes the request with an
    /// id-bound `ParseError` response instead of a bare channel error.
    #[tokio::test(flavor = "multi_thread")]
    async fn client_falls_back_when_discover_gets_a_non_json_4xx() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(listener, log.clone(), DiscoverReply::Html400));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        client
            .connect()
            .await
            .expect("fallback connect must succeed");
        assert!(client.is_legacy_peer(), "peer must be marked legacy");

        let tools = client
            .list_tools(None)
            .await
            .expect("tools/list must work after the fallback");
        assert!(tools.tools.is_empty());
    }

    /// A server that answers `server/discover` *successfully* but with a
    /// protocol version this build does not support has committed to the
    /// 2026-07-28 path -- the client must surface the real version error, not
    /// mark the peer legacy and chase `initialize` on a cancelled
    /// transport.
    #[tokio::test(flavor = "multi_thread")]
    async fn successful_discover_with_unsupported_version_does_not_fall_back() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(
            listener,
            log.clone(),
            DiscoverReply::UnsupportedVersion,
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        let err = client
            .connect()
            .await
            .expect_err("an unsupported 2026-07-28 version must fail the connect");
        assert!(
            err.to_string().contains("but the client speaks"),
            "the real version error must surface, got: {err}"
        );
        assert!(
            !client.is_legacy_peer(),
            "a successful discovery must never mark the peer legacy"
        );
        assert!(
            client.handler.is_none() && client.cancellation_token.is_none(),
            "a failed negotiation must not leave the transport running"
        );

        let log = log
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        assert!(
            !log.iter().any(|r| r.contains("\"method\":\"initialize\"")),
            "no initialize fallback may be attempted after successful discovery"
        );
    }

    /// `notifications/roots/list_changed` is gone from MCP 2026-07-28, but a
    /// peer reached through the fallback negotiated `roots.listChanged` on the
    /// legacy protocol -- and would otherwise hold a stale root list forever.
    // Roots are deprecated under MCP 2026-07-28 -- which is exactly why the
    // fallback still owes the legacy peer this notification.
    #[allow(deprecated)]
    #[tokio::test(flavor = "multi_thread")]
    async fn roots_changes_are_pushed_to_a_fallback_peer() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(
            listener,
            log.clone(),
            DiscoverReply::MethodNotFound,
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_roots(|roots| roots.with_list_changed())
                .with_timeout(std::time::Duration::from_secs(5))
        });

        client.connect().await.expect("fallback must connect");
        assert!(client.is_legacy_peer(), "the mock is a legacy server");

        client.add_root("file:///tmp/project", "Project");

        // The push is fire-and-forget through a spawned task.
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            let seen = log
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .iter()
                .any(|r| r.contains("notifications/roots/list_changed"));
            if seen {
                break;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "a legacy peer must be told its root list changed"
            );
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
    }

    /// A protected 2026-07-28 endpoint replying `401` with a non-JSON body must
    /// surface the authentication failure, not be mistaken for a legacy
    /// peer -- otherwise the client silently drops the 2026-07-28 headers and
    /// retries `initialize`, masking the real cause.
    #[tokio::test(flavor = "multi_thread")]
    async fn unauthorized_discover_does_not_fall_back() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(
            listener,
            log.clone(),
            DiscoverReply::Unauthorized401,
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        let err = client
            .connect()
            .await
            .expect_err("an unauthenticated connect must fail");
        assert!(
            err.to_string().contains("401"),
            "the HTTP status must be carried through, got: {err}"
        );
        assert!(
            !client.is_legacy_peer(),
            "an auth failure must never mark the peer legacy"
        );

        let log = log
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        assert!(
            !log.iter().any(|r| r.contains("\"method\":\"initialize\"")),
            "no initialize fallback may be attempted after an auth failure"
        );
    }

    /// An upstream outage (reverse proxy `503`, rate limit, gateway
    /// timeout) says nothing about the peer's protocol generation: the
    /// failure must surface instead of being read as "legacy" and retried
    /// as `initialize` into the very same outage.
    #[tokio::test(flavor = "multi_thread")]
    async fn upstream_failure_during_discover_does_not_fall_back() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let log = Arc::new(Mutex::new(Vec::<String>::new()));
        tokio::spawn(serve_legacy(
            listener,
            log.clone(),
            DiscoverReply::Unavailable503,
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        let err = client
            .connect()
            .await
            .expect_err("an upstream outage must fail the connect");
        assert!(
            err.to_string().contains("503"),
            "the upstream status must surface, got: {err}"
        );
        assert!(
            !client.is_legacy_peer(),
            "an upstream outage must never mark the peer legacy"
        );

        let log = log
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        assert!(
            !log.iter().any(|r| r.contains("\"method\":\"initialize\"")),
            "no initialize fallback may be attempted on an upstream failure"
        );
    }

    /// A `with_mcp_version` legacy override must not leak into the 2026-07-28
    /// expectation: `server/discover` still expects the 2026-07-28 version, the
    /// override only selects the fallback's negotiated legacy version.
    #[test]
    fn legacy_version_override_keeps_the_latest_expectation() {
        let client = Client::new().with_options(|opt| opt.with_mcp_version("2025-06-18"));
        assert_eq!(
            client.expected_protocol_ver(),
            crate::LATEST_PROTOCOL_VERSION
        );

        client.options.peer_mode.set_legacy();
        assert_eq!(client.expected_protocol_ver(), "2025-06-18");
    }
}

/// The other half of the "Done when": against a 2026-07-28 server the client
/// keeps using `server/discover` (no fallback).
#[cfg(all(
    test,
    feature = "http-client",
    feature = "http-server-volga",
    not(feature = "legacy-spec")
))]
mod roundtrip_tests {
    use super::*;

    #[tokio::test(flavor = "multi_thread")]
    async fn client_discovers_a_2026_07_28_server() {
        let mut app = crate::App::new()
            .with_options(|opt| opt.with_http(|http| http.bind("127.0.0.1:39817")));
        app.map_tool("echo", |name: String| async move { name });
        tokio::spawn(app.run());

        // Wait until the server socket actually accepts connections --
        // a fixed sleep is not enough on loaded CI machines.
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
        loop {
            match tokio::net::TcpStream::connect("127.0.0.1:39817").await {
                Ok(_) => break,
                Err(_) if tokio::time::Instant::now() < deadline => {
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await
                }
                Err(err) => panic!("2026-07-28 server never became reachable: {err}"),
            }
        }

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind("127.0.0.1:39817"))
                .with_timeout(std::time::Duration::from_secs(5))
        });

        client.connect().await.expect("discover must succeed");
        assert!(!client.is_legacy_peer(), "2026-07-28 peers never fall back");

        let tools = client.list_tools(None).await.expect("tools/list");
        assert_eq!(tools.tools.len(), 1);
        assert_eq!(tools.tools[0].name, "echo");

        // `serverInfo` left `DiscoverResult`: it now rides in every result's
        // `_meta`, and the MRTR send path -- which every 2026-07-28 request
        // takes -- is what has to pick it up.
        assert!(
            client.server_info.is_some(),
            "the server identifies itself in every result's `_meta`"
        );
    }
}

/// What the client will mirror into `Mcp-Param-*` headers is decided by the
/// current listing and nothing else: a tool the server no longer designates --
/// or no longer lists at all -- must stop sending its argument in a header.
#[cfg(all(test, feature = "http-client", not(feature = "legacy-spec")))]
mod param_header_registry_tests {
    use super::*;

    fn listing(tools: serde_json::Value) -> ListToolsResult {
        serde_json::from_value(serde_json::json!({ "tools": tools })).expect("valid listing")
    }

    fn annotated(name: &str) -> serde_json::Value {
        serde_json::json!({
            "name": name,
            "inputSchema": {
                "type": "object",
                "properties": { "region": { "type": "string", "x-mcp-header": "Region" } }
            }
        })
    }

    fn plain(name: &str) -> serde_json::Value {
        serde_json::json!({
            "name": name,
            "inputSchema": { "type": "object", "properties": { "q": { "type": "string" } } }
        })
    }

    #[test]
    fn a_fresh_listing_forgets_a_tool_it_no_longer_lists() {
        let mut client = Client::new();

        let mut first = listing(serde_json::json!([annotated("search")]));
        client.register_param_headers(&mut first, true);
        assert!(client.options.param_headers.contains_key("search"));

        // The tool is gone from the refreshed listing -- a later direct
        // `call_tool("search", ..)` must not keep mirroring its argument.
        let mut second = listing(serde_json::json!([plain("other")]));
        client.register_param_headers(&mut second, true);
        assert!(client.options.param_headers.is_empty());
    }

    #[test]
    fn a_dropped_annotation_is_forgotten_on_the_same_tool() {
        let mut client = Client::new();

        let mut first = listing(serde_json::json!([annotated("search")]));
        client.register_param_headers(&mut first, true);

        let mut second = listing(serde_json::json!([plain("search")]));
        client.register_param_headers(&mut second, true);
        assert!(client.options.param_headers.is_empty());
    }

    /// Where a listing came from does not change what it binds: a batched
    /// `tools/list` registers and filters exactly as a direct one does.
    #[test]
    fn a_batched_listing_registers_and_filters() {
        let mut client = Client::new();

        let mut resp = Response::success(
            RequestId::Number(1),
            serde_json::json!({
                "tools": [
                    annotated("search"),
                    {
                        "name": "broken",
                        "inputSchema": {
                            "type": "object",
                            "properties": {
                                "p": { "type": "array", "items": { "x-mcp-header": "P" } }
                            }
                        }
                    }
                ]
            }),
        );
        client.register_batched_tools(&mut resp);

        assert!(client.options.param_headers.contains_key("search"));
        assert!(!client.options.param_headers.contains_key("broken"));

        // The caller must not be handed a tool the client refuses to call.
        let Response::Ok(ok) = &resp else {
            panic!("a successful listing")
        };
        let tools = ok.result["tools"].as_array().expect("tools array");
        assert_eq!(tools.len(), 1);
        assert_eq!(tools[0]["name"], "search");
    }

    /// A slot that is not a listing is the caller's to interpret.
    #[test]
    fn a_non_listing_response_is_left_alone() {
        let mut client = Client::new();
        let mut resp = Response::success(
            RequestId::Number(1),
            serde_json::json!({ "content": [{ "type": "text", "text": "hi" }] }),
        );
        let before = match &resp {
            Response::Ok(ok) => ok.result.clone(),
            Response::Err(_) => panic!("a successful response"),
        };

        client.register_batched_tools(&mut resp);

        let Response::Ok(ok) = &resp else {
            panic!("a successful response")
        };
        assert_eq!(ok.result, before);
        assert!(client.options.param_headers.is_empty());
    }

    #[test]
    fn later_pages_accumulate_onto_the_traversal() {
        let mut client = Client::new();

        let mut page1 = listing(serde_json::json!([annotated("search")]));
        client.register_param_headers(&mut page1, true);

        // A tool absent from page two was not withdrawn, only listed earlier.
        let mut page2 = listing(serde_json::json!([annotated("lookup")]));
        client.register_param_headers(&mut page2, false);

        assert!(client.options.param_headers.contains_key("search"));
        assert!(client.options.param_headers.contains_key("lookup"));
    }

    #[test]
    fn an_invalid_definition_drops_the_tool_and_its_registration() {
        let mut client = Client::new();

        let mut first = listing(serde_json::json!([annotated("search")]));
        client.register_param_headers(&mut first, true);

        // Same tool, now annotated somewhere the client cannot reach.
        let mut second = listing(serde_json::json!([{
            "name": "search",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "region": { "type": "array", "items": { "x-mcp-header": "Region" } }
                }
            }
        }]));
        client.register_param_headers(&mut second, true);

        assert!(second.tools.is_empty(), "a malformed tool is not callable");
        assert!(client.options.param_headers.is_empty());
    }

    fn call(name: &str) -> Request {
        Request::new(
            Some(RequestId::Number(1)),
            crate::types::tool::commands::CALL,
            Some(serde_json::json!({ "name": name, "arguments": {} })),
        )
    }

    /// Hiding the name from the listing is not enough on its own: a caller
    /// holding it from anywhere else would otherwise reach the tool with none
    /// of the headers its declaration asked for.
    #[test]
    fn a_rejected_tool_cannot_be_called_by_name() {
        let mut client = Client::new();

        let mut listed = listing(serde_json::json!([
            annotated("search"),
            {
                "name": "broken",
                "inputSchema": {
                    "type": "object",
                    "properties": { "p": { "type": "array", "items": { "x-mcp-header": "P" } } }
                }
            }
        ]));
        client.register_param_headers(&mut listed, true);

        let err = client
            .blocked_tool_error(&call("broken"))
            .expect("a rejected tool is refused");
        assert_eq!(err.code, ErrorCode::InvalidParams);
        assert!(client.blocked_tool_error(&call("search")).is_none());
        // Only `tools/call` names a tool.
        assert!(
            client
                .blocked_tool_error(&Request::new(
                    Some(RequestId::Number(1)),
                    crate::types::tool::commands::LIST,
                    Some(serde_json::json!({ "name": "broken" })),
                ))
                .is_none()
        );
    }

    /// The block follows the listing: a definition the server fixed -- or
    /// withdrew altogether -- is no longer the one being refused.
    #[test]
    fn a_fresh_listing_lifts_the_block() {
        let mut client = Client::new();

        let mut first = listing(serde_json::json!([{
            "name": "broken",
            "inputSchema": {
                "type": "object",
                "properties": { "p": { "type": "array", "items": { "x-mcp-header": "P" } } }
            }
        }]));
        client.register_param_headers(&mut first, true);
        assert!(client.blocked_tool_error(&call("broken")).is_some());

        let mut fixed = listing(serde_json::json!([annotated("broken")]));
        client.register_param_headers(&mut fixed, true);
        assert!(client.blocked_tool_error(&call("broken")).is_none());

        client.register_param_headers(&mut first, true);
        let mut gone = listing(serde_json::json!([plain("other")]));
        client.register_param_headers(&mut gone, true);
        assert!(client.blocked_tool_error(&call("broken")).is_none());
    }
}

/// Task support in a 2026-07-28 build means one thing: the peer advertised the
/// extension *and* stayed on the 2026-07-28 protocol. Generations do not mix
/// for tasks -- run a `legacy-spec` build against a legacy server.
#[cfg(all(test, feature = "tasks", not(feature = "legacy-spec")))]
mod fallback_tasks_capability_tests {
    use super::*;
    use crate::types::ServerCapabilities;
    use serde_json::json;

    fn client_with_capabilities(caps: serde_json::Value) -> Client {
        let mut client = Client::new();
        client.server_capabilities =
            Some(serde_json::from_value::<ServerCapabilities>(caps).expect("valid capabilities"));
        client
    }

    /// The legacy top-level `tasks` field can only reach this build from a
    /// fallback peer, whose task protocol it does not speak. Reading it would
    /// promise support that ends in 2026-07-28 messages the peer cannot read.
    #[test]
    fn a_legacy_top_level_tasks_capability_is_not_support() {
        let client = client_with_capabilities(json!({
            "tools": {},
            "tasks": { "requests": { "tools": { "call": {} } } }
        }));
        assert!(!client.is_server_supports_tasks());
        assert!(!client.is_server_support_call_tool_with_tasks());
    }

    /// And a peer that advertised the 2026-07-28 extension but then negotiated
    /// the legacy protocol is no different.
    #[test]
    fn a_fallback_peer_reports_no_task_support() {
        let client = client_with_capabilities(json!({
            "tools": {},
            "extensions": {
                "io.modelcontextprotocol/tasks": { "requests": { "tools": { "call": {} } } }
            }
        }));
        assert!(client.is_server_supports_tasks());

        client.options.peer_mode.set_legacy();
        assert!(!client.is_server_supports_tasks());
        assert!(!client.is_server_support_call_tool_with_tasks());
    }

    #[test]
    fn the_extension_tasks_capability_resolves() {
        let client = client_with_capabilities(json!({
            "tools": {},
            "extensions": {
                "io.modelcontextprotocol/tasks": { "requests": { "tools": { "call": {} } } }
            }
        }));
        assert!(client.is_server_supports_tasks());
    }

    #[test]
    fn no_tasks_capability_resolves_to_none() {
        let client = client_with_capabilities(json!({ "tools": {} }));
        assert!(!client.is_server_supports_tasks());
    }
}

/// Establishing a subscription against a peer that misbehaves: answering
/// instead of acknowledging, or acknowledging more than was asked for. Driven
/// by a raw-HTTP mock, since a real neva server produces neither.
#[cfg(all(test, feature = "http-client", not(feature = "legacy-spec")))]
mod listen_rejection_tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::{TcpListener, TcpStream};

    const REASON: &str = "subscriptions are disabled here";

    async fn read_request(stream: &mut TcpStream) -> Option<(String, String)> {
        let mut buf = Vec::new();
        let mut tmp = [0u8; 2048];
        let header_end = loop {
            let n = stream.read(&mut tmp).await.ok()?;
            if n == 0 {
                return None;
            }
            buf.extend_from_slice(&tmp[..n]);
            if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                break pos + 4;
            }
            if buf.len() > 65536 {
                return None;
            }
        };
        let head = String::from_utf8_lossy(&buf[..header_end]).to_string();
        let content_length = head
            .lines()
            .find_map(|l| {
                l.to_ascii_lowercase()
                    .strip_prefix("content-length:")
                    .map(|v| v.trim().parse::<usize>().ok())
            })
            .flatten()
            .unwrap_or(0);
        while buf.len() < header_end + content_length {
            let n = stream.read(&mut tmp).await.ok()?;
            if n == 0 {
                return None;
            }
            buf.extend_from_slice(&tmp[..n]);
        }
        let body =
            String::from_utf8_lossy(&buf[header_end..header_end + content_length]).to_string();
        Some((head, body))
    }

    async fn write_json(stream: &mut TcpStream, body: &str) {
        let resp = format!(
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: keep-alive\r\n\r\n{body}",
            body.len()
        );
        let _ = stream.write_all(resp.as_bytes()).await;
    }

    /// Answers `server/discover` normally, then rejects `subscriptions/listen`
    /// with a JSON-RPC error and never sends an acknowledgment.
    async fn serve_rejecting(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    let reply = if method == crate::commands::DISCOVER {
                        serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": { "tools": { "listChanged": true } }
                            }
                        })
                    } else {
                        serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "error": { "code": -32600, "message": REASON }
                        })
                    };
                    write_json(&mut stream, &reply.to_string()).await;
                }
            });
        }
    }

    /// A cancelled subscription is never answered, so nothing completes its
    /// request slot -- and those slots carry no TTL, because a subscription may
    /// legitimately stay open for hours. Whoever gives up on one has to release
    /// it, or a client that opens and cancels subscriptions in a loop grows the
    /// pending queue by one entry per cycle.
    #[tokio::test(flavor = "multi_thread")]
    async fn cancelling_releases_the_request_slot() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_offside_notification(listener));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");

        let queued = |client: &Client| client.handler.as_ref().expect("connected").pending().len();

        let idle = queued(&client);

        let mut subscription = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect("listen");
        assert_eq!(
            queued(&client),
            idle + 1,
            "the live subscription holds one slot"
        );

        subscription.cancel().await.expect("cancel");
        assert_eq!(
            queued(&client),
            idle,
            "cancelling must release the subscription's slot"
        );
    }

    /// A cancel written immediately behind a listen -- which is exactly what a
    /// dropped establishment writes -- has to find the stream it names. The
    /// abort handle is registered by the connection loop rather than by the
    /// task it spawns, so the two are ordered by the wire, not by the
    /// scheduler.
    ///
    /// A single worker on purpose: it pins the interleaving this is about.
    /// Both messages are queued before the connection loop runs, so it reads
    /// the cancel on the turn right after spawning the listen, while that task
    /// has had no chance to run. (One worker rather than `current_thread`
    /// because connecting uses `block_in_place`, which the current-thread
    /// runtime refuses.)
    #[tokio::test(flavor = "multi_thread")]
    async fn a_cancel_queued_behind_a_listen_still_closes_the_stream() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let opened = Arc::new(AtomicBool::new(false));
        let hung_up = Arc::new(AtomicBool::new(false));
        tokio::spawn(serve_orphan_watch(
            listener,
            opened.clone(),
            hung_up.clone(),
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(30))
        });
        client.connect().await.expect("connect");

        use crate::transport::Sender as _;

        let id = RequestId::Number(99);
        let mut sender = client.handler.as_ref().expect("connected").sender();

        let listen = Request::new(
            Some(id.clone()),
            crate::types::subscription::commands::LISTEN,
            Some(SubscriptionsListenRequestParams::new(
                crate::types::SubscriptionFilter::new().with_tools_changed(),
            )),
        );
        sender.send(listen.into()).await.expect("send listen");
        sender
            .send(subscription::cancelled(&id).into())
            .await
            .expect("send cancel");

        // The abort may land before the POST is even written, so what is
        // asserted is the outcome rather than one particular mechanism: the
        // peer must not be left holding this listen open. Without the fix the
        // cancel finds no handle, the request goes out, and the stream drains
        // for as long as the connection lives.
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(3);
        while !hung_up.load(AtomicOrdering::SeqCst) && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }

        assert!(
            !opened.load(AtomicOrdering::SeqCst) || hung_up.load(AtomicOrdering::SeqCst),
            "a cancel arriving right behind its listen left the peer holding an orphaned stream"
        );
    }

    /// Answers `server/discover`, then holds any other request open and reports
    /// both that it saw one and whether the client ever hung up on it.
    async fn serve_orphan_watch(
        listener: TcpListener,
        opened: Arc<AtomicBool>,
        hung_up: Arc<AtomicBool>,
    ) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            let (opened, hung_up) = (opened.clone(), hung_up.clone());
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };

                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": { "tools": { "listChanged": true } }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    // Anything else -- the cancel notification travels on its
                    // own `POST` -- is answered and forgotten; only the listen
                    // is the stream this test is about.
                    if method != crate::types::subscription::commands::LISTEN {
                        let _ = stream
                            .write_all(
                                b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n",
                            )
                            .await;
                        continue;
                    }

                    opened.store(true, AtomicOrdering::SeqCst);

                    let mut probe = [0u8; 1];
                    if let Ok(Ok(0)) = tokio::time::timeout(
                        std::time::Duration::from_secs(3),
                        stream.read(&mut probe),
                    )
                    .await
                    {
                        hung_up.store(true, AtomicOrdering::SeqCst);
                    }

                    return;
                }
            });
        }
    }

    /// A caller who drops the `listen` future -- an outer `timeout`, a lost
    /// `select!` branch -- runs none of the error paths inside it, so the
    /// bookkeeping and the peer's stream would be left behind by an
    /// establishment that never returned anything to end them with.
    #[tokio::test(flavor = "multi_thread")]
    async fn dropping_the_listen_future_abandons_the_subscription() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let closed = Arc::new(AtomicBool::new(false));
        tokio::spawn(serve_stalled_listen(listener, closed.clone()));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                // Far longer than this test waits: the establishment must be
                // ended by the dropped future, not by `listen`'s own timeout.
                .with_timeout(std::time::Duration::from_secs(30))
        });
        client.connect().await.expect("connect");

        let queued = |client: &Client| client.handler.as_ref().expect("connected").pending().len();
        let idle = queued(&client);

        // The caller gives up on its own schedule and never sees a result.
        assert!(
            tokio::time::timeout(
                std::time::Duration::from_millis(300),
                client.listen(crate::types::SubscriptionFilter::new().with_tools_changed()),
            )
            .await
            .is_err(),
            "the peer never acknowledges, so the outer timeout must fire"
        );

        assert_eq!(
            queued(&client),
            idle,
            "a dropped establishment must release its request slot"
        );

        // And it has to reach the wire too: the peer is still holding the
        // listen request open, waiting for someone who is no longer there.
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        while !closed.load(AtomicOrdering::SeqCst) && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            closed.load(AtomicOrdering::SeqCst),
            "a dropped establishment must close the stream it opened"
        );
    }

    /// A cancel that never reached the wire ended nothing -- the connection
    /// did. Reporting `Cancelled` for it would tell the caller a deliberate
    /// stop happened where a connection loss did, and reconnect logic keys on
    /// exactly that difference.
    #[tokio::test(flavor = "multi_thread")]
    async fn cancelling_after_a_disconnect_reports_abrupt() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_offside_notification(listener));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");

        let mut subscription = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect("listen");

        client.disconnect().await.expect("disconnect");

        assert!(
            subscription.cancel().await.is_err(),
            "a cancel has nowhere to go once the transport is gone"
        );

        let ended = tokio::time::timeout(std::time::Duration::from_secs(2), subscription.closed())
            .await
            .expect("closed() must not hang after a failed cancel");
        assert!(
            matches!(ended, crate::client::SubscriptionEnd::Abrupt),
            "got {ended:?}"
        );
    }

    /// The same slot must come back when the handle is simply dropped.
    #[tokio::test(flavor = "multi_thread")]
    async fn dropping_the_handle_releases_the_request_slot() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_offside_notification(listener));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");

        let queued = |client: &Client| client.handler.as_ref().expect("connected").pending().len();

        let idle = queued(&client);
        {
            let _subscription = client
                .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
                .await
                .expect("listen");
            assert_eq!(queued(&client), idle + 1);
        }

        assert_eq!(
            queued(&client),
            idle,
            "dropping the handle must release the subscription's slot"
        );
    }

    /// Acknowledges exactly what was asked for, then sends a tagged
    /// notification of a category outside it.
    async fn serve_offside_notification(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": {
                                    "tools": { "listChanged": true },
                                    "prompts": { "listChanged": true }
                                }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n";
                    let _ = stream.write_all(head.as_bytes()).await;

                    // A correct acknowledgment: tools only, exactly as asked.
                    let ack = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": "notifications/subscriptions/acknowledged",
                        "params": {
                            "notifications": { "toolsListChanged": true },
                            "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id }
                        }
                    });
                    let _ = stream
                        .write_all(format!("data: {ack}\n\n").as_bytes())
                        .await;

                    // ...then a category the filter never selected.
                    let offside = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": crate::types::prompt::commands::LIST_CHANGED,
                        "params": { "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id } }
                    });
                    let _ = stream
                        .write_all(format!("data: {offside}\n\n").as_bytes())
                        .await;

                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    return;
                }
            });
        }
    }

    /// A valid acknowledgment is a promise about the whole stream, not just its
    /// first message. A peer that keeps it and then sends an off-filter
    /// notification anyway must not reach the client's handlers -- they are
    /// global and know nothing about which subscription a message came from.
    #[tokio::test(flavor = "multi_thread")]
    async fn notifications_outside_the_acknowledged_filter_are_dropped() {
        use std::sync::atomic::AtomicUsize;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_offside_notification(listener));

        let tools = Arc::new(AtomicUsize::new(0));
        let prompts = Arc::new(AtomicUsize::new(0));
        let (tools_seen, prompts_seen) = (tools.clone(), prompts.clone());

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");
        client.subscribe(crate::types::tool::commands::LIST_CHANGED, move |_| {
            let seen = tools_seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });
        client.subscribe(crate::types::prompt::commands::LIST_CHANGED, move |_| {
            let seen = prompts_seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });

        let _subscription = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect("listen");

        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        assert_eq!(
            prompts.load(AtomicOrdering::SeqCst),
            0,
            "a notification outside the acknowledged filter must be dropped"
        );
        assert_eq!(
            tools.load(AtomicOrdering::SeqCst),
            0,
            "and the in-filter categories are unaffected (none were sent)"
        );
    }

    /// Answers `server/discover` normally, then puts a correctly tagged,
    /// in-filter notification on the stream *ahead* of the acknowledgment.
    async fn serve_notification_before_ack(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": { "tools": { "listChanged": true } }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n";
                    let _ = stream.write_all(head.as_bytes()).await;

                    // Correctly tagged, squarely inside what was requested --
                    // and out of order.
                    let early = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": crate::types::tool::commands::LIST_CHANGED,
                        "params": { "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id } }
                    });
                    let _ = stream
                        .write_all(format!("data: {early}\n\n").as_bytes())
                        .await;

                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    return;
                }
            });
        }
    }

    /// The acknowledgment comes first or the subscription is not established.
    /// A peer that streams before acknowledging is streaming from a
    /// subscription `listen` goes on to report as failed -- its events must not
    /// have reached the handlers in the meantime.
    #[tokio::test(flavor = "multi_thread")]
    async fn notifications_before_the_acknowledgment_are_dropped() {
        use std::sync::atomic::AtomicUsize;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_notification_before_ack(listener));

        let tools = Arc::new(AtomicUsize::new(0));
        let seen = tools.clone();

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                // Bounded, because the peer never acknowledges and
                // establishment has to end by timing out -- but not tight: the
                // same budget covers the `connect` above, which shares this
                // client, and a loaded machine makes a short one flake.
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");
        client.subscribe(crate::types::tool::commands::LIST_CHANGED, move |_| {
            let seen = seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });

        client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect_err("a peer that never acknowledges must not establish");

        assert_eq!(
            tools.load(AtomicOrdering::SeqCst),
            0,
            "a notification sent before the acknowledgment must be dropped"
        );
    }

    /// Answers `server/discover` normally, acknowledges the subscription, then
    /// pushes a subscribable notification with no subscription id on it.
    async fn serve_untagged_notification(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": { "tools": { "listChanged": true } }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n";
                    let _ = stream.write_all(head.as_bytes()).await;

                    let ack = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": "notifications/subscriptions/acknowledged",
                        "params": {
                            "notifications": { "toolsListChanged": true },
                            "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id }
                        }
                    });
                    let _ = stream
                        .write_all(format!("data: {ack}\n\n").as_bytes())
                        .await;

                    // In the accepted filter, but with nothing tying it to the
                    // subscription that accepted it.
                    let untagged = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": crate::types::tool::commands::LIST_CHANGED
                    });
                    let _ = stream
                        .write_all(format!("data: {untagged}\n\n").as_bytes())
                        .await;

                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    return;
                }
            });
        }
    }

    /// Under MCP 2026-07-28 a subscribable notification travels on a
    /// subscription and nowhere else. One that arrives without a subscription
    /// id has nothing to check it against, so it cannot be admitted just
    /// because the client happens to have asked for that category somewhere.
    #[tokio::test(flavor = "multi_thread")]
    async fn untagged_subscribable_notifications_are_dropped() {
        use std::sync::atomic::AtomicUsize;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_untagged_notification(listener));

        let tools = Arc::new(AtomicUsize::new(0));
        let seen = tools.clone();

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");
        client.subscribe(crate::types::tool::commands::LIST_CHANGED, move |_| {
            let seen = seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });

        let _subscription = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect("listen");

        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        assert_eq!(
            tools.load(AtomicOrdering::SeqCst),
            0,
            "a subscription-only notification with no subscription id must be dropped"
        );
    }

    /// A transport that dies while `listen` is still waiting for its
    /// acknowledgment is a lost connection, not a peer that would not
    /// acknowledge -- and callers act on those differently.
    #[tokio::test(flavor = "multi_thread")]
    async fn listen_reports_a_lost_transport_as_a_connection_error() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_stalled_listen(
            listener,
            Arc::new(AtomicBool::new(false)),
        ));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                // Long enough that a timeout cannot be what ends the wait.
                .with_timeout(std::time::Duration::from_secs(30))
        });
        client.connect().await.expect("connect");

        let token = client.handler.as_ref().expect("connected").cancellation();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
            token.cancel();
        });

        let err = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.listen(crate::types::SubscriptionFilter::new().with_tools_changed()),
        )
        .await
        .expect("a dead transport must not be waited out")
        .expect_err("a dead transport cannot establish a subscription");

        assert_eq!(err.code, ErrorCode::InternalError, "got {err:?}");
    }

    /// Answers `server/discover` normally, then delivers the whole
    /// subscription -- acknowledgment, an in-filter notification and an
    /// off-filter one -- inside a single JSON-RPC batch.
    async fn serve_batched_frames(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };

            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };

                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": {
                                    "tools": { "listChanged": true },
                                    "prompts": { "listChanged": true }
                                }
                            }
                        });

                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n";
                    let _ = stream.write_all(head.as_bytes()).await;

                    let batch = serde_json::json!([
                        {
                            "jsonrpc": "2.0",
                            "method": "notifications/subscriptions/acknowledged",
                            "params": {
                                "notifications": { "toolsListChanged": true },
                                "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id }
                            }
                        },
                        {
                            "jsonrpc": "2.0",
                            "method": crate::types::tool::commands::LIST_CHANGED,
                            "params": { "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id } }
                        },
                        {
                            "jsonrpc": "2.0",
                            "method": crate::types::prompt::commands::LIST_CHANGED,
                            "params": { "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id } }
                        }
                    ]);
                    let _ = stream
                        .write_all(format!("data: {batch}\n\n").as_bytes())
                        .await;

                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    return;
                }
            });
        }
    }

    /// Batching is a framing choice of the peer's. An acknowledgment sent that
    /// way still has to establish the subscription, and a tagged notification
    /// sent that way still has to face the filter it was accepted under.
    #[tokio::test(flavor = "multi_thread")]
    async fn batched_subscription_frames_go_through_the_same_gate() {
        use std::sync::atomic::AtomicUsize;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_batched_frames(listener));

        let tools = Arc::new(AtomicUsize::new(0));
        let prompts = Arc::new(AtomicUsize::new(0));
        let (tools_seen, prompts_seen) = (tools.clone(), prompts.clone());

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(30))
        });
        client.connect().await.expect("connect");
        client.subscribe(crate::types::tool::commands::LIST_CHANGED, move |_| {
            let seen = tools_seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });
        client.subscribe(crate::types::prompt::commands::LIST_CHANGED, move |_| {
            let seen = prompts_seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });

        // A batched acknowledgment has to resolve `listen`, not leave it
        // waiting out the establishment timeout.
        let _subscription = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect("a batched acknowledgment must establish the subscription");

        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        assert_eq!(
            tools.load(AtomicOrdering::SeqCst),
            1,
            "an in-filter batched notification must be delivered"
        );
        assert_eq!(
            prompts.load(AtomicOrdering::SeqCst),
            0,
            "an off-filter batched notification must be dropped"
        );
    }

    /// Answers `server/discover` normally, then acknowledges the subscription
    /// with a *broader* filter than the client asked for.
    async fn serve_overbroad_ack(listener: TcpListener) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": {
                                    "tools": { "listChanged": true },
                                    "prompts": { "listChanged": true }
                                }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    // The client asked for tools only; claim prompts as well.
                    let ack = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": "notifications/subscriptions/acknowledged",
                        "params": {
                            "notifications": {
                                "toolsListChanged": true,
                                "promptsListChanged": true
                            },
                            "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id }
                        }
                    });
                    let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n";
                    let _ = stream.write_all(head.as_bytes()).await;
                    let _ = stream
                        .write_all(format!("data: {ack}\n\n").as_bytes())
                        .await;

                    // Straight behind it, a notification squarely inside what
                    // the client *did* ask for. The acknowledgment is on its way
                    // to being rejected, so this must not reach the handlers
                    // either -- intersecting the filter alone would let it.
                    let in_filter = serde_json::json!({
                        "jsonrpc": "2.0",
                        "method": crate::types::tool::commands::LIST_CHANGED,
                        "params": { "_meta": { crate::types::SUBSCRIPTION_ID_KEY: id } }
                    });
                    let _ = stream
                        .write_all(format!("data: {in_filter}\n\n").as_bytes())
                        .await;

                    // Hold the stream open, the way a real subscription would.
                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    return;
                }
            });
        }
    }

    /// An acknowledgment may narrow the requested filter -- that is what it is
    /// for -- but never widen it. Notifications reach the client's global
    /// handlers with no per-subscription filtering, so accepting an overbroad
    /// acknowledgment would deliver events this call never asked for.
    #[tokio::test(flavor = "multi_thread")]
    async fn listen_rejects_an_overbroad_acknowledgment() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_overbroad_ack(listener));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");

        let tools = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let seen = tools.clone();
        client.subscribe(crate::types::tool::commands::LIST_CHANGED, move |_| {
            let seen = seen.clone();
            async move {
                seen.fetch_add(1, AtomicOrdering::SeqCst);
            }
        });

        let err = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect_err("an overbroad acknowledgment must be rejected");

        let reported = format!("{err:?}");
        assert!(
            reported.contains("broader"),
            "the rejection must name the cause, got: {reported}"
        );

        // Nothing may have been delivered on the strength of an acknowledgment
        // this call refuses -- not even a category it did ask for.
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        assert_eq!(
            tools.load(AtomicOrdering::SeqCst),
            0,
            "a subscription `listen` rejects must deliver nothing"
        );
    }

    /// Answers `server/discover` normally, then accepts the listen `POST` and
    /// never replies at all -- headers included. Records whether that
    /// connection was closed by the peer.
    async fn serve_stalled_listen(listener: TcpListener, closed: Arc<AtomicBool>) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            let closed = closed.clone();
            tokio::spawn(async move {
                loop {
                    let Some((_head, body)) = read_request(&mut stream).await else {
                        return;
                    };
                    let msg: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
                    let id = msg.get("id").cloned().unwrap_or(serde_json::Value::Null);
                    let method = msg
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or_default();

                    if method == crate::commands::DISCOVER {
                        let reply = serde_json::json!({
                            "jsonrpc": "2.0", "id": id,
                            "result": {
                                "supportedVersions": [crate::LATEST_PROTOCOL_VERSION],
                                "capabilities": { "tools": { "listChanged": true } }
                            }
                        });
                        write_json(&mut stream, &reply.to_string()).await;
                        continue;
                    }

                    // Sit on the request without sending so much as a status
                    // line, and watch for the client hanging up.
                    let mut probe = [0u8; 1];
                    let hung_up = tokio::time::timeout(
                        std::time::Duration::from_secs(10),
                        stream.read(&mut probe),
                    )
                    .await
                    .is_ok_and(|read| matches!(read, Ok(0)));
                    if hung_up {
                        closed.store(true, AtomicOrdering::SeqCst);
                    }
                    return;
                }
            });
        }
    }

    /// A subscription can be cancelled while the peer is still sitting on the
    /// response headers -- establishment timing out is exactly that. The abort
    /// handle has to exist by then, or the cancel finds nothing to close and the
    /// task starts draining an orphaned stream once the headers finally arrive.
    #[tokio::test(flavor = "multi_thread")]
    async fn listen_closes_a_stalled_request_on_timeout() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let closed = Arc::new(AtomicBool::new(false));
        tokio::spawn(serve_stalled_listen(listener, closed.clone()));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                // Bounded, because establishment has to give up on its own --
                // but not tight: this budget covers the `connect` before the
                // subscription as well, and a loaded machine makes a short one
                // flake. The assertion below waits longer still.
                .with_timeout(std::time::Duration::from_secs(5))
        });
        client.connect().await.expect("connect");

        client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect_err("a stalled subscription must not establish");

        // Giving up has to reach the wire: the peer sees the connection go away
        // instead of holding a request nobody is waiting for.
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        while !closed.load(AtomicOrdering::SeqCst) && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            closed.load(AtomicOrdering::SeqCst),
            "abandoning an unacknowledged subscription must close its request"
        );
    }

    /// A rejected `subscriptions/listen` must surface the server's own error
    /// immediately. Waiting only on the acknowledgment would sit out the full
    /// request timeout and report *that* instead -- the peer's reason lost, and
    /// the caller blocked for no reason.
    #[tokio::test(flavor = "multi_thread")]
    async fn listen_surfaces_an_immediate_rejection() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(serve_rejecting(listener));

        let mut client = Client::new().with_options(|opt| {
            opt.with_http(|http| http.bind(addr.to_string()))
                // Generous on purpose: a timeout would be unmistakable below.
                .with_timeout(std::time::Duration::from_secs(30))
        });
        client.connect().await.expect("connect");

        let started = tokio::time::Instant::now();
        let err = client
            .listen(crate::types::SubscriptionFilter::new().with_tools_changed())
            .await
            .expect_err("a rejected subscription must fail");

        let reported = format!("{err:?}");
        assert!(
            reported.contains(REASON),
            "the server's own error must survive, got: {reported}"
        );
        assert!(
            started.elapsed() < std::time::Duration::from_secs(5),
            "the rejection must not wait out the request timeout"
        );
    }
}