tower_mcp/

context.rs

//! Request context for MCP handlers
//!
//! Provides progress reporting, cancellation support, and client request capabilities
//! for long-running operations.
//!
//! # Example
//!
//! ```rust,ignore
//! use tower_mcp::context::RequestContext;
//!
//! async fn long_running_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
//!     for i in 0..100 {
//!         // Check if cancelled
//!         if ctx.is_cancelled() {
//!             return Err(Error::tool("Operation cancelled"));
//!         }
//!
//!         // Report progress
//!         ctx.report_progress(i as f64, Some(100.0), Some("Processing...")).await;
//!
//!         do_work(i).await;
//!     }
//!     Ok(CallToolResult::text("Done!"))
//! }
//! ```
//!
//! # Sampling (LLM requests to client)
//!
//! ```rust,ignore
//! use tower_mcp::context::RequestContext;
//! use tower_mcp::{CreateMessageParams, SamplingMessage};
//!
//! async fn ai_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
//!     // Request LLM completion from the client
//!     let params = CreateMessageParams::new(
//!         vec![SamplingMessage::user("Summarize this text...")],
//!         500,
//!     );
//!
//!     let result = ctx.sample(params).await?;
//!     Ok(CallToolResult::text(format!("Summary: {:?}", result.content)))
//! }
//! ```
//!
//! # Elicitation (requesting user input)
//!
//! ```rust,ignore
//! use tower_mcp::context::RequestContext;
//! use tower_mcp::{ElicitFormParams, ElicitFormSchema, ElicitMode, ElicitAction};
//!
//! async fn interactive_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
//!     // Request user input via form
//!     let params = ElicitFormParams {
//!         mode: Some(ElicitMode::Form),
//!         message: "Please provide additional details".to_string(),
//!         requested_schema: ElicitFormSchema::new()
//!             .string_field("name", Some("Your name"), true)
//!             .number_field("age", Some("Your age"), false),
//!         meta: None,
//!     };
//!
//!     let result = ctx.elicit_form(params).await?;
//!     if result.action == ElicitAction::Accept {
//!         // Use the form data
//!         Ok(CallToolResult::text(format!("Got: {:?}", result.content)))
//!     } else {
//!         Ok(CallToolResult::text("User declined"))
//!     }
//! }
//! ```
//!
//! # Stateless mode: per-request metadata (`stateless` feature)
//!
//! With the 2026-07-28 protocol, clients do not run an initialize handshake.
//! Instead, every request carries the client's protocol version, identity, and
//! capabilities in a `_meta` object. JSON-RPC transports extract these fields
//! and stash them as a [`StatelessRequestMeta`](crate::stateless::StatelessRequestMeta)
//! extension on the [`RequestContext`], accessible via
//! [`ctx.per_request_meta()`](RequestContext::per_request_meta).
//!
//! `per_request_meta()` returns `Some` when:
//! - The `stateless` feature is compiled in, AND
//! - The request was dispatched by a JSON-RPC transport, AND
//! - The request's `_meta` contained at least one recognized field.
//!
//! It returns `None` for 2025-11-25 session-based requests and when the request
//! carried no modern protocol metadata.
//!
//! The [`StatelessRequestMeta`](crate::stateless::StatelessRequestMeta) struct
//! provides:
//!
//! - `protocol_version` -- the `io.modelcontextprotocol/protocolVersion` field
//! - `client_info` -- the `io.modelcontextprotocol/clientInfo` field (name, version)
//! - `client_capabilities` -- the `io.modelcontextprotocol/clientCapabilities` field
//! - `log_level` -- optional per-request log level override
//! - `progress_token` -- optional progress token for progress notifications
//!
//! ```rust,ignore
//! // Requires feature = ["stateless"]
//! use tower_mcp::context::RequestContext;
//!
//! async fn my_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
//!     if let Some(meta) = ctx.per_request_meta() {
//!         // Available for 2026-07-28 clients on JSON-RPC transports
//!         if let Some(ref info) = meta.client_info {
//!             tracing::info!(client = %info.name, version = %info.version, "request from");
//!         }
//!         if let Some(ref version) = meta.protocol_version {
//!             tracing::debug!(protocol_version = %version);
//!         }
//!     }
//!     Ok(CallToolResult::text("ok"))
//! }
//! ```

use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, RwLock};

use async_trait::async_trait;
use tokio::sync::mpsc;

use crate::error::{Error, Result};
use crate::protocol::{
    CallToolResult, CancelTaskParams, CreateMessageParams, CreateMessageResult, ElicitFormParams,
    ElicitRequestParams, ElicitResult, ElicitUrlParams, GetTaskInfoParams, GetTaskResultParams,
    ListTasksParams, ListTasksResult, LogLevel, LoggingMessageParams, ProgressParams,
    ProgressToken, RequestId, TaskObject, TaskStatus,
};

/// A notification to be sent to the client
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum ServerNotification {
    /// Progress update for a request
    Progress(ProgressParams),
    /// Log message notification
    LogMessage(LoggingMessageParams),
    /// A subscribed resource has been updated
    ResourceUpdated {
        /// The URI of the updated resource
        uri: String,
    },
    /// The list of available resources has changed
    ResourcesListChanged,
    /// The list of available tools has changed
    ToolsListChanged,
    /// The list of available prompts has changed
    PromptsListChanged,
    /// Task status has changed, in the legacy flat shape.
    TaskStatusChanged(crate::protocol::TaskStatusParams),
    /// Task status has changed, in the final SEP-2663 shape.
    ///
    /// Carries the complete status-discriminated task, identical to what
    /// `tasks/get` would have returned at that moment. Delivered only on
    /// `subscriptions/listen` streams that named this task ID.
    FinalTaskStatusChanged(crate::tasks::TaskStatusNotificationParams),
}

/// Sender for server notifications
pub type NotificationSender = mpsc::Sender<ServerNotification>;

/// Receiver for server notifications
pub type NotificationReceiver = mpsc::Receiver<ServerNotification>;

/// Create a new notification channel
pub fn notification_channel(buffer: usize) -> (NotificationSender, NotificationReceiver) {
    mpsc::channel(buffer)
}

// =============================================================================
// Client Requests (Server -> Client)
// =============================================================================

/// Trait for sending requests from server to client
///
/// This enables bidirectional communication where the server can request
/// actions from the client, such as sampling (LLM requests), elicitation
/// (user input requests), and task polling (per SEP-1686).
#[async_trait]
pub trait ClientRequester: Send + Sync {
    /// Send a sampling request to the client
    ///
    /// Returns the LLM completion result from the client.
    async fn sample(&self, params: CreateMessageParams) -> Result<CreateMessageResult>;

    /// Send an elicitation request to the client
    ///
    /// This requests user input from the client. The request can be either
    /// form-based (structured input) or URL-based (redirect to external URL).
    ///
    /// Returns the elicitation result with the user's action and any submitted data.
    async fn elicit(&self, params: ElicitRequestParams) -> Result<ElicitResult>;

    /// Send a generic JSON-RPC request to the client.
    ///
    /// Used by typed helpers ([`RequestContext::get_task_info`] etc.) to
    /// dispatch arbitrary request methods. The default implementation returns
    /// an error so existing custom implementations of this trait keep
    /// compiling; they only need to override this if they want to support
    /// methods beyond `sample` and `elicit`.
    async fn request(
        &self,
        method: String,
        params: serde_json::Value,
    ) -> Result<serde_json::Value> {
        let _ = (method, params);
        Err(Error::Internal(
            "ClientRequester does not support arbitrary requests".to_string(),
        ))
    }
}

/// A clonable handle to a client requester
pub type ClientRequesterHandle = Arc<dyn ClientRequester>;

/// Outgoing request to be sent to the client
#[derive(Debug)]
pub struct OutgoingRequest {
    /// The JSON-RPC request ID
    pub id: RequestId,
    /// The method name
    pub method: String,
    /// The request parameters as JSON
    pub params: serde_json::Value,
    /// Channel to send the response back
    pub response_tx: tokio::sync::oneshot::Sender<Result<serde_json::Value>>,
}

/// Sender for outgoing requests to the client
pub type OutgoingRequestSender = mpsc::Sender<OutgoingRequest>;

/// Receiver for outgoing requests (used by transport)
pub type OutgoingRequestReceiver = mpsc::Receiver<OutgoingRequest>;

/// Create a new outgoing request channel
pub fn outgoing_request_channel(buffer: usize) -> (OutgoingRequestSender, OutgoingRequestReceiver) {
    mpsc::channel(buffer)
}

/// A client requester implementation that sends requests through a channel
#[derive(Clone)]
pub struct ChannelClientRequester {
    request_tx: OutgoingRequestSender,
    next_id: Arc<AtomicI64>,
}

impl ChannelClientRequester {
    /// Create a new channel-based client requester
    pub fn new(request_tx: OutgoingRequestSender) -> Self {
        Self {
            request_tx,
            next_id: Arc::new(AtomicI64::new(1)),
        }
    }

    /// Create a requester that draws IDs from a transport-owned allocator.
    ///
    /// HTTP uses one allocator per session while giving each originating POST
    /// its own request channel. This keeps server-to-client request IDs unique
    /// without allowing those requests to escape onto an unrelated SSE stream.
    #[cfg(feature = "http")]
    pub(crate) fn with_id_allocator(
        request_tx: OutgoingRequestSender,
        next_id: Arc<AtomicI64>,
    ) -> Self {
        Self {
            request_tx,
            next_id,
        }
    }

    fn next_request_id(&self) -> RequestId {
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        RequestId::Number(id)
    }
}

impl ChannelClientRequester {
    async fn dispatch(&self, method: &str, params: serde_json::Value) -> Result<serde_json::Value> {
        let id = self.next_request_id();
        let (response_tx, response_rx) = tokio::sync::oneshot::channel();

        let request = OutgoingRequest {
            id,
            method: method.to_string(),
            params,
            response_tx,
        };

        self.request_tx
            .send(request)
            .await
            .map_err(|_| Error::Internal("Failed to send request: channel closed".to_string()))?;

        response_rx.await.map_err(|_| {
            Error::Internal("Failed to receive response: channel closed".to_string())
        })?
    }
}

#[async_trait]
impl ClientRequester for ChannelClientRequester {
    async fn sample(&self, params: CreateMessageParams) -> Result<CreateMessageResult> {
        let params_json = serde_json::to_value(&params)
            .map_err(|e| Error::Internal(format!("Failed to serialize params: {}", e)))?;
        let response = self.dispatch("sampling/createMessage", params_json).await?;
        serde_json::from_value(response)
            .map_err(|e| Error::Internal(format!("Failed to deserialize response: {}", e)))
    }

    async fn elicit(&self, params: ElicitRequestParams) -> Result<ElicitResult> {
        let params_json = serde_json::to_value(&params)
            .map_err(|e| Error::Internal(format!("Failed to serialize params: {}", e)))?;
        let response = self.dispatch("elicitation/create", params_json).await?;
        serde_json::from_value(response)
            .map_err(|e| Error::Internal(format!("Failed to deserialize response: {}", e)))
    }

    async fn request(
        &self,
        method: String,
        params: serde_json::Value,
    ) -> Result<serde_json::Value> {
        self.dispatch(&method, params).await
    }
}

/// Context for a request, providing progress, cancellation, and client request support
#[derive(Clone)]
pub struct RequestContext {
    /// The request ID
    request_id: RequestId,
    /// Progress token (if provided by client)
    progress_token: Option<ProgressToken>,
    /// Cancellation signal for this request
    cancellation: tokio_util::sync::CancellationToken,
    /// Channel for sending notifications
    notification_tx: Option<NotificationSender>,
    /// Handle for sending requests to the client (for sampling, etc.)
    client_requester: Option<ClientRequesterHandle>,
    /// Extensions for passing data from router/middleware to handlers
    extensions: Arc<Extensions>,
    /// Minimum log level set by the client (shared with router for dynamic updates)
    min_log_level: Option<Arc<RwLock<LogLevel>>>,
}

/// Type-erased extensions map for passing data to handlers.
///
/// Extensions allow router-level state and middleware-injected data to flow
/// to tool handlers via the `Extension<T>` extractor.
#[derive(Clone, Default)]
pub struct Extensions {
    map: std::collections::HashMap<std::any::TypeId, Arc<dyn std::any::Any + Send + Sync>>,
}

impl Extensions {
    /// Create an empty extensions map.
    pub fn new() -> Self {
        Self::default()
    }

    /// Insert a value into the extensions map.
    ///
    /// If a value of the same type already exists, it is replaced.
    pub fn insert<T: Send + Sync + 'static>(&mut self, val: T) {
        self.map.insert(std::any::TypeId::of::<T>(), Arc::new(val));
    }

    /// Get a reference to a value in the extensions map.
    ///
    /// Returns `None` if no value of the given type has been inserted.
    pub fn get<T: Send + Sync + 'static>(&self) -> Option<&T> {
        self.map
            .get(&std::any::TypeId::of::<T>())
            .and_then(|val| val.downcast_ref::<T>())
    }

    /// Check if the extensions map contains a value of the given type.
    pub fn contains<T: Send + Sync + 'static>(&self) -> bool {
        self.map.contains_key(&std::any::TypeId::of::<T>())
    }

    /// Merge another extensions map into this one.
    ///
    /// Values from `other` will overwrite existing values of the same type.
    pub fn merge(&mut self, other: &Extensions) {
        for (k, v) in &other.map {
            self.map.insert(*k, v.clone());
        }
    }

    /// Returns the number of entries in the extensions map.
    pub fn len(&self) -> usize {
        self.map.len()
    }

    /// Returns `true` if the extensions map contains no entries.
    pub fn is_empty(&self) -> bool {
        self.map.is_empty()
    }
}

impl std::fmt::Debug for Extensions {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Extensions")
            .field("len", &self.map.len())
            .finish()
    }
}

impl std::fmt::Debug for RequestContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RequestContext")
            .field("request_id", &self.request_id)
            .field("progress_token", &self.progress_token)
            .field("cancelled", &self.cancellation.is_cancelled())
            .finish()
    }
}

impl RequestContext {
    /// Create a new request context
    pub fn new(request_id: RequestId) -> Self {
        Self {
            request_id,
            progress_token: None,
            cancellation: tokio_util::sync::CancellationToken::new(),
            notification_tx: None,
            client_requester: None,
            extensions: Arc::new(Extensions::new()),
            min_log_level: None,
        }
    }

    /// Set the progress token
    pub fn with_progress_token(mut self, token: ProgressToken) -> Self {
        self.progress_token = Some(token);
        self
    }

    /// Set the notification sender
    pub fn with_notification_sender(mut self, tx: NotificationSender) -> Self {
        self.notification_tx = Some(tx);
        self
    }

    /// Set the minimum log level for filtering outgoing log notifications
    ///
    /// This is shared with the router so that `logging/setLevel` updates
    /// are immediately visible to all request contexts.
    pub fn with_min_log_level(mut self, level: Arc<RwLock<LogLevel>>) -> Self {
        self.min_log_level = Some(level);
        self
    }

    /// Set the client requester for server-to-client requests
    pub fn with_client_requester(mut self, requester: ClientRequesterHandle) -> Self {
        self.client_requester = Some(requester);
        self
    }

    /// Set the extensions for this request context.
    ///
    /// Extensions allow router-level state and middleware data to flow to handlers.
    pub fn with_extensions(mut self, extensions: Arc<Extensions>) -> Self {
        self.extensions = extensions;
        self
    }

    /// Get a reference to a value from the extensions map.
    ///
    /// Returns `None` if no value of the given type has been inserted.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// #[derive(Clone)]
    /// struct CurrentUser { id: String }
    ///
    /// // In a handler:
    /// if let Some(user) = ctx.extension::<CurrentUser>() {
    ///     println!("User: {}", user.id);
    /// }
    /// ```
    pub fn extension<T: Send + Sync + 'static>(&self) -> Option<&T> {
        self.extensions.get::<T>()
    }

    /// Protocol extensions declared by both the client and server.
    ///
    /// Unknown or one-sided declarations are not included. The returned view
    /// preserves each peer's settings object for extension-specific policy.
    pub fn negotiated_extensions(&self) -> Option<&crate::NegotiatedExtensions> {
        self.extension()
    }

    /// Get a mutable reference to the extensions.
    ///
    /// This allows middleware to insert data that handlers can access via
    /// the `Extension<T>` extractor.
    pub fn extensions_mut(&mut self) -> &mut Extensions {
        Arc::make_mut(&mut self.extensions)
    }

    /// Get a reference to the extensions.
    pub fn extensions(&self) -> &Extensions {
        &self.extensions
    }

    /// SEP-2575 per-request `_meta` (protocol version, client info, client
    /// capabilities, log level) if the transport extracted it.
    ///
    /// Returns `Some` for 2026-07-28 clients on JSON-RPC transports when the
    /// request carried a `_meta` object with recognized fields. Returns `None`
    /// when:
    /// - The request had no `_meta` field, or
    /// - The transport does not use [`crate::jsonrpc::JsonRpcService`], or
    /// - The `stateless` feature is not compiled in.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// async fn my_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
    ///     if let Some(meta) = ctx.per_request_meta() {
    ///         // protocol_version, client_info, client_capabilities are all Option<_>
    ///         if let Some(ref version) = meta.protocol_version {
    ///             tracing::debug!(protocol_version = %version);
    ///         }
    ///         if let Some(ref info) = meta.client_info {
    ///             tracing::info!(client = %info.name, version = %info.version);
    ///         }
    ///     }
    ///     Ok(CallToolResult::text("ok"))
    /// }
    /// ```
    #[cfg(feature = "stateless")]
    pub fn per_request_meta(&self) -> Option<&crate::stateless::StatelessRequestMeta> {
        self.extension::<crate::stateless::StatelessRequestMeta>()
    }

    /// SEP-2322 continuation values supplied by the client on this attempt.
    #[cfg(feature = "stateless")]
    pub fn mrtr(&self) -> Option<&crate::mrtr::MrtrRequest> {
        self.extension::<crate::mrtr::MrtrRequest>()
    }

    /// Client responses from the prior MRTR round, if any.
    #[cfg(feature = "stateless")]
    pub fn input_responses(&self) -> Option<&crate::protocol::InputResponses> {
        self.mrtr()
            .and_then(crate::mrtr::MrtrRequest::input_responses)
    }

    /// Opaque request state echoed by the client, if any.
    #[cfg(feature = "stateless")]
    pub fn request_state(&self) -> Option<&str> {
        self.mrtr()
            .and_then(crate::mrtr::MrtrRequest::request_state)
    }

    /// Router-configured request-state codec shared by this handler.
    #[cfg(feature = "stateless")]
    pub fn request_state_codec(&self) -> Option<&crate::mrtr::RequestStateCodec> {
        self.extension::<crate::mrtr::RequestStateCodec>()
    }

    /// Get the request ID
    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    /// Get the progress token (if any)
    pub fn progress_token(&self) -> Option<&ProgressToken> {
        self.progress_token.as_ref()
    }

    /// Check if the request has been cancelled
    pub fn is_cancelled(&self) -> bool {
        self.cancellation.is_cancelled()
    }

    /// Mark the request as cancelled
    pub fn cancel(&self) {
        self.cancellation.cancel();
    }

    /// Wait until the request is cancelled.
    ///
    /// Completes when [`cancel`](Self::cancel) is called -- by a
    /// `notifications/cancelled` message, or by the transport when the
    /// client disconnects before the response is delivered (HTTP
    /// stateless mode). Useful in `tokio::select!` to abandon work early:
    ///
    /// ```rust,ignore
    /// tokio::select! {
    ///     result = do_work() => { /* ... */ }
    ///     _ = ctx.cancelled() => return Err(Error::tool("cancelled")),
    /// }
    /// ```
    pub async fn cancelled(&self) {
        self.cancellation.cancelled().await
    }

    /// Get a cancellation token that can be shared
    pub fn cancellation_token(&self) -> CancellationToken {
        CancellationToken {
            inner: self.cancellation.clone(),
        }
    }

    /// Replace this context's cancellation source with an existing token.
    ///
    /// Used by transports to link a request's lifetime to an external
    /// signal (e.g. client disconnect on the HTTP stateless path). After
    /// this call, `is_cancelled()`, `cancelled()`, and tokens returned by
    /// [`cancellation_token`](Self::cancellation_token) all observe the
    /// given token.
    pub fn with_cancellation_token(mut self, token: CancellationToken) -> Self {
        self.cancellation = token.inner;
        self
    }

    /// Report progress to the client
    ///
    /// This is a no-op if no progress token was provided or no notification sender is configured.
    pub async fn report_progress(&self, progress: f64, total: Option<f64>, message: Option<&str>) {
        let Some(token) = &self.progress_token else {
            return;
        };
        let Some(tx) = &self.notification_tx else {
            return;
        };

        let params = ProgressParams {
            progress_token: token.clone(),
            progress,
            total,
            message: message.map(|s| s.to_string()),
            meta: None,
        };

        // Best effort - don't block if channel is full
        let _ = tx.try_send(ServerNotification::Progress(params));
    }

    /// Report progress synchronously (non-async version)
    ///
    /// This is a no-op if no progress token was provided or no notification sender is configured.
    pub fn report_progress_sync(&self, progress: f64, total: Option<f64>, message: Option<&str>) {
        let Some(token) = &self.progress_token else {
            return;
        };
        let Some(tx) = &self.notification_tx else {
            return;
        };

        let params = ProgressParams {
            progress_token: token.clone(),
            progress,
            total,
            message: message.map(|s| s.to_string()),
            meta: None,
        };

        let _ = tx.try_send(ServerNotification::Progress(params));
    }

    /// Notify subscribed clients that the tool list changed.
    pub fn notify_tools_list_changed(&self) -> bool {
        self.notification_tx
            .as_ref()
            .is_some_and(|tx| tx.try_send(ServerNotification::ToolsListChanged).is_ok())
    }

    /// Notify subscribed clients that the prompt list changed.
    pub fn notify_prompts_list_changed(&self) -> bool {
        self.notification_tx
            .as_ref()
            .is_some_and(|tx| tx.try_send(ServerNotification::PromptsListChanged).is_ok())
    }

    /// Notify subscribed clients that the resource list changed.
    pub fn notify_resources_list_changed(&self) -> bool {
        self.notification_tx.as_ref().is_some_and(|tx| {
            tx.try_send(ServerNotification::ResourcesListChanged)
                .is_ok()
        })
    }

    /// Notify subscribed clients that one resource changed.
    pub fn notify_resource_updated(&self, uri: impl Into<String>) -> bool {
        self.notification_tx.as_ref().is_some_and(|tx| {
            tx.try_send(ServerNotification::ResourceUpdated { uri: uri.into() })
                .is_ok()
        })
    }

    /// Notify the client that a final-protocol task changed status.
    ///
    /// This is useful when a handler drives a task transition through a custom
    /// [`crate::TaskStore`] path rather than through the router directly.
    pub fn notify_task_status_changed(
        &self,
        params: crate::tasks::TaskStatusNotificationParams,
    ) -> bool {
        self.notification_tx.as_ref().is_some_and(|tx| {
            tx.try_send(ServerNotification::FinalTaskStatusChanged(params))
                .is_ok()
        })
    }

    /// Send a log message notification to the client
    ///
    /// This is a no-op if no notification sender is configured.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use tower_mcp::protocol::{LoggingMessageParams, LogLevel};
    ///
    /// async fn my_tool(ctx: RequestContext) {
    ///     ctx.send_log(
    ///         LoggingMessageParams::new(LogLevel::Info, serde_json::json!("Processing..."))
    ///             .with_logger("my-tool")
    ///     );
    /// }
    /// ```
    pub fn send_log(&self, params: LoggingMessageParams) {
        let Some(tx) = &self.notification_tx else {
            return;
        };

        // The final protocol removed logging/setLevel. Log delivery is instead
        // authorized per request: no logLevel means no log notifications.
        #[cfg(feature = "stateless")]
        if let Some(meta) = self.per_request_meta()
            && meta.protocol_version.as_deref()
                == Some(crate::protocol::PROTOCOL_VERSION_2026_07_28)
        {
            let Some(request_level) = meta.log_level else {
                return;
            };
            let request_level = match request_level {
                crate::stateless::LogLevel::Debug => LogLevel::Debug,
                crate::stateless::LogLevel::Info => LogLevel::Info,
                crate::stateless::LogLevel::Notice => LogLevel::Notice,
                crate::stateless::LogLevel::Warning => LogLevel::Warning,
                crate::stateless::LogLevel::Error => LogLevel::Error,
                crate::stateless::LogLevel::Critical => LogLevel::Critical,
                crate::stateless::LogLevel::Alert => LogLevel::Alert,
                crate::stateless::LogLevel::Emergency => LogLevel::Emergency,
            };
            if params.level > request_level {
                return;
            }
            let _ = tx.try_send(ServerNotification::LogMessage(params));
            return;
        }

        // Filter by minimum log level set via logging/setLevel
        // LogLevel derives Ord with Emergency < Alert < ... < Debug,
        // so a message passes if its severity is at least the minimum
        // (i.e., its ordinal is <= the minimum level's ordinal).
        if let Some(min_level) = &self.min_log_level
            && let Ok(min) = min_level.read()
            && params.level > *min
        {
            return;
        }

        let _ = tx.try_send(ServerNotification::LogMessage(params));
    }

    /// Check if sampling is available
    ///
    /// Returns true if a client requester is configured and the transport
    /// supports bidirectional communication.
    pub fn can_sample(&self) -> bool {
        self.client_requester.is_some()
    }

    /// Request an LLM completion from the client
    ///
    /// This sends a `sampling/createMessage` request to the client and waits
    /// for the response. The client is expected to forward this to an LLM
    /// and return the result.
    ///
    /// Returns an error if sampling is not available (no client requester configured).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use tower_mcp::{CreateMessageParams, SamplingMessage};
    ///
    /// async fn my_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
    ///     let params = CreateMessageParams::new(
    ///         vec![SamplingMessage::user("Summarize: ...")],
    ///         500,
    ///     );
    ///
    ///     let result = ctx.sample(params).await?;
    ///     Ok(CallToolResult::text(format!("{:?}", result.content)))
    /// }
    /// ```
    pub async fn sample(&self, params: CreateMessageParams) -> Result<CreateMessageResult> {
        let requester = self.client_requester.as_ref().ok_or_else(|| {
            Error::Internal("Sampling not available: no client requester configured".to_string())
        })?;

        requester.sample(params).await
    }

    /// Check if elicitation is available
    ///
    /// Returns true if a client requester is configured and the transport
    /// supports bidirectional communication. Note that this only checks if
    /// the mechanism is available, not whether the client supports elicitation.
    pub fn can_elicit(&self) -> bool {
        self.client_requester.is_some()
    }

    /// Request user input via a form from the client
    ///
    /// This sends an `elicitation/create` request to the client with a form schema.
    /// The client renders the form to the user and returns their response.
    ///
    /// Returns an error if elicitation is not available (no client requester configured).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use tower_mcp::{ElicitFormParams, ElicitFormSchema, ElicitMode, ElicitAction};
    ///
    /// async fn my_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
    ///     let params = ElicitFormParams {
    ///         mode: Some(ElicitMode::Form),
    ///         message: "Please enter your details".to_string(),
    ///         requested_schema: ElicitFormSchema::new()
    ///             .string_field("name", Some("Your name"), true),
    ///         meta: None,
    ///     };
    ///
    ///     let result = ctx.elicit_form(params).await?;
    ///     match result.action {
    ///         ElicitAction::Accept => {
    ///             // Use result.content
    ///             Ok(CallToolResult::text("Got your input!"))
    ///         }
    ///         _ => Ok(CallToolResult::text("User declined"))
    ///     }
    /// }
    /// ```
    pub async fn elicit_form(&self, params: ElicitFormParams) -> Result<ElicitResult> {
        let requester = self.client_requester.as_ref().ok_or_else(|| {
            Error::Internal("Elicitation not available: no client requester configured".to_string())
        })?;

        requester.elicit(ElicitRequestParams::Form(params)).await
    }

    /// Request user input via URL redirect from the client
    ///
    /// This sends an `elicitation/create` request to the client with a URL.
    /// The client directs the user to the URL for out-of-band input collection.
    /// The server receives the result via a callback notification.
    ///
    /// Returns an error if elicitation is not available (no client requester configured).
    ///
    /// **Protocol note:** `ElicitUrlParams::elicitation_id` and the callback
    /// notification it correlates with are a 2025-11-25-and-earlier pattern.
    /// The final 2026-07-28 schema removes both in favor of MRTR (SEP-2322).
    /// Use an MRTR-capable tool, prompt, or resource handler there; the client
    /// learns the outcome by retrying the original request instead of receiving
    /// a completion notification.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use tower_mcp::{ElicitUrlParams, ElicitMode, ElicitAction};
    ///
    /// async fn my_tool(ctx: RequestContext, input: MyInput) -> Result<CallToolResult> {
    ///     let params = ElicitUrlParams {
    ///         mode: Some(ElicitMode::Url),
    ///         elicitation_id: "unique-id-123".to_string(),
    ///         message: "Please authorize via the link".to_string(),
    ///         url: "https://example.com/auth?id=unique-id-123".to_string(),
    ///         meta: None,
    ///     };
    ///
    ///     let result = ctx.elicit_url(params).await?;
    ///     match result.action {
    ///         ElicitAction::Accept => Ok(CallToolResult::text("Authorization complete!")),
    ///         _ => Ok(CallToolResult::text("Authorization cancelled"))
    ///     }
    /// }
    /// ```
    pub async fn elicit_url(&self, params: ElicitUrlParams) -> Result<ElicitResult> {
        let requester = self.client_requester.as_ref().ok_or_else(|| {
            Error::Internal("Elicitation not available: no client requester configured".to_string())
        })?;

        requester.elicit(ElicitRequestParams::Url(params)).await
    }

    /// Request simple confirmation from the user.
    ///
    /// This is a convenience method for simple yes/no confirmation dialogs.
    /// It creates an elicitation form with a single boolean "confirm" field
    /// and returns `true` if the user accepts, `false` otherwise.
    ///
    /// Returns an error if elicitation is not available (no client requester configured).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use tower_mcp::{RequestContext, CallToolResult};
    ///
    /// async fn delete_item(ctx: RequestContext) -> Result<CallToolResult> {
    ///     let confirmed = ctx.confirm("Are you sure you want to delete this item?").await?;
    ///     if confirmed {
    ///         // Perform deletion
    ///         Ok(CallToolResult::text("Item deleted"))
    ///     } else {
    ///         Ok(CallToolResult::text("Deletion cancelled"))
    ///     }
    /// }
    /// ```
    pub async fn confirm(&self, message: impl Into<String>) -> Result<bool> {
        use crate::protocol::{ElicitAction, ElicitFormParams, ElicitFormSchema, ElicitMode};

        let params = ElicitFormParams {
            mode: Some(ElicitMode::Form),
            message: message.into(),
            requested_schema: ElicitFormSchema::new().boolean_field_with_default(
                "confirm",
                Some("Confirm this action"),
                true,
                false,
            ),
            meta: None,
        };

        let result = self.elicit_form(params).await?;
        Ok(result.action == ElicitAction::Accept)
    }

    /// List tasks tracked by the connected client (legacy SEP-1686).
    ///
    /// Sends a `tasks/list` request to the client and returns the result.
    /// Pass `Some(status)` to filter to a single status, or `None` for all
    /// tasks. Pagination is exposed via [`ListTasksResult::next_cursor`];
    /// use [`request_raw`](Self::request_raw) for cursor-driven calls.
    ///
    /// Returns an error if no client requester is configured or the client
    /// does not advertise task support.
    #[deprecated(
        since = "0.13.0",
        note = "final SEP-2663 removes tasks/list; a conforming peer answers \
                MethodNotFound (-32601). Only useful against legacy SEP-1686 \
                clients."
    )]
    pub async fn list_tasks(&self, status: Option<TaskStatus>) -> Result<ListTasksResult> {
        let params = ListTasksParams {
            status,
            cursor: None,
            meta: None,
        };
        let value = self
            .request_raw("tasks/list", serde_json::to_value(&params)?)
            .await?;
        serde_json::from_value(value)
            .map_err(|e| Error::Internal(format!("Failed to deserialize tasks/list: {e}")))
    }

    /// Fetch metadata for a single task tracked by the client (SEP-1686).
    ///
    /// Sends a `tasks/get` request and returns the task object, including
    /// the current status, timestamps, and TTL.
    pub async fn get_task_info(&self, task_id: impl Into<String>) -> Result<TaskObject> {
        let params = GetTaskInfoParams {
            task_id: task_id.into(),
            meta: None,
        };
        let value = self
            .request_raw("tasks/get", serde_json::to_value(&params)?)
            .await?;
        serde_json::from_value(value)
            .map_err(|e| Error::Internal(format!("Failed to deserialize tasks/get: {e}")))
    }

    /// Fetch the terminal result for a task tracked by the client (legacy
    /// SEP-1686).
    ///
    /// Sends a `tasks/result` request. The client is expected to block until
    /// the task reaches a terminal state and then return the underlying
    /// `CallToolResult`. For long-running tasks, prefer polling with
    /// [`get_task_info`](Self::get_task_info) and only call this once the
    /// status is terminal.
    #[deprecated(
        since = "0.13.0",
        note = "final SEP-2663 removes tasks/result (results are inlined in \
                the tasks/get DetailedTask); a conforming peer answers \
                MethodNotFound (-32601). Only useful against legacy SEP-1686 \
                clients."
    )]
    pub async fn get_task_result(&self, task_id: impl Into<String>) -> Result<CallToolResult> {
        let params = GetTaskResultParams {
            task_id: task_id.into(),
            meta: None,
        };
        let value = self
            .request_raw("tasks/result", serde_json::to_value(&params)?)
            .await?;
        serde_json::from_value(value)
            .map_err(|e| Error::Internal(format!("Failed to deserialize tasks/result: {e}")))
    }

    /// Cancel a task tracked by the client.
    ///
    /// Sends a `tasks/cancel` request. Per final SEP-2663 the acknowledgment
    /// is an empty result and the observable task status is polled via
    /// [`get_task_info`](Self::get_task_info); the ack body is discarded, so
    /// this also tolerates legacy SEP-1686 peers that return the task object.
    pub async fn cancel_task(
        &self,
        task_id: impl Into<String>,
        reason: Option<String>,
    ) -> Result<()> {
        let params = CancelTaskParams {
            task_id: task_id.into(),
            reason,
            meta: None,
        };
        self.request_raw("tasks/cancel", serde_json::to_value(&params)?)
            .await?;
        Ok(())
    }

    /// Send an arbitrary JSON-RPC request to the client.
    ///
    /// Escape hatch for methods not covered by the typed helpers (e.g. when
    /// a `tasks/list` cursor needs to be passed). Most callers should prefer
    /// the typed methods.
    pub async fn request_raw(
        &self,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value> {
        let requester = self.client_requester.as_ref().ok_or_else(|| {
            Error::Internal(
                "Client request not available: no client requester configured".to_string(),
            )
        })?;
        requester.request(method.to_string(), params).await
    }
}

/// A token that can be used to check for, wait on, or request cancellation
///
/// Cloned tokens share the same underlying signal: cancelling any clone
/// cancels them all. Backed by [`tokio_util::sync::CancellationToken`],
/// so cancellation can also be awaited via [`cancelled`](Self::cancelled).
#[derive(Clone, Debug, Default)]
pub struct CancellationToken {
    inner: tokio_util::sync::CancellationToken,
}

impl CancellationToken {
    /// Create a new, un-cancelled token
    pub fn new() -> Self {
        Self::default()
    }

    /// Check if cancellation has been requested
    pub fn is_cancelled(&self) -> bool {
        self.inner.is_cancelled()
    }

    /// Request cancellation
    pub fn cancel(&self) {
        self.inner.cancel();
    }

    /// Wait until cancellation is requested
    ///
    /// Completes immediately if the token is already cancelled.
    pub async fn cancelled(&self) {
        self.inner.cancelled().await
    }
}

/// Builder for creating request contexts
#[derive(Default)]
pub struct RequestContextBuilder {
    request_id: Option<RequestId>,
    progress_token: Option<ProgressToken>,
    notification_tx: Option<NotificationSender>,
    client_requester: Option<ClientRequesterHandle>,
    min_log_level: Option<Arc<RwLock<LogLevel>>>,
}

impl RequestContextBuilder {
    /// Create a new builder
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the request ID
    pub fn request_id(mut self, id: RequestId) -> Self {
        self.request_id = Some(id);
        self
    }

    /// Set the progress token
    pub fn progress_token(mut self, token: ProgressToken) -> Self {
        self.progress_token = Some(token);
        self
    }

    /// Set the notification sender
    pub fn notification_sender(mut self, tx: NotificationSender) -> Self {
        self.notification_tx = Some(tx);
        self
    }

    /// Set the client requester for server-to-client requests
    pub fn client_requester(mut self, requester: ClientRequesterHandle) -> Self {
        self.client_requester = Some(requester);
        self
    }

    /// Set the minimum log level for filtering
    pub fn min_log_level(mut self, level: Arc<RwLock<LogLevel>>) -> Self {
        self.min_log_level = Some(level);
        self
    }

    /// Build the request context
    ///
    /// Panics if request_id is not set.
    pub fn build(self) -> RequestContext {
        let mut ctx = RequestContext::new(self.request_id.expect("request_id is required"));
        if let Some(token) = self.progress_token {
            ctx = ctx.with_progress_token(token);
        }
        if let Some(tx) = self.notification_tx {
            ctx = ctx.with_notification_sender(tx);
        }
        if let Some(requester) = self.client_requester {
            ctx = ctx.with_client_requester(requester);
        }
        if let Some(level) = self.min_log_level {
            ctx = ctx.with_min_log_level(level);
        }
        ctx
    }
}

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

    #[test]
    fn test_cancellation() {
        let ctx = RequestContext::new(RequestId::Number(1));
        assert!(!ctx.is_cancelled());

        let token = ctx.cancellation_token();
        assert!(!token.is_cancelled());

        ctx.cancel();
        assert!(ctx.is_cancelled());
        assert!(token.is_cancelled());
    }

    #[tokio::test]
    async fn test_progress_reporting() {
        let (tx, mut rx) = notification_channel(10);

        let ctx = RequestContext::new(RequestId::Number(1))
            .with_progress_token(ProgressToken::Number(42))
            .with_notification_sender(tx);

        ctx.report_progress(50.0, Some(100.0), Some("Halfway"))
            .await;

        let notification = rx.recv().await.unwrap();
        match notification {
            ServerNotification::Progress(params) => {
                assert_eq!(params.progress, 50.0);
                assert_eq!(params.total, Some(100.0));
                assert_eq!(params.message.as_deref(), Some("Halfway"));
            }
            _ => panic!("Expected Progress notification"),
        }
    }

    #[tokio::test]
    async fn test_progress_no_token() {
        let (tx, mut rx) = notification_channel(10);

        // No progress token - should be a no-op
        let ctx = RequestContext::new(RequestId::Number(1)).with_notification_sender(tx);

        ctx.report_progress(50.0, Some(100.0), None).await;

        // Channel should be empty
        assert!(rx.try_recv().is_err());
    }

    #[test]
    fn test_builder() {
        let (tx, _rx) = notification_channel(10);

        let ctx = RequestContextBuilder::new()
            .request_id(RequestId::String("req-1".to_string()))
            .progress_token(ProgressToken::String("prog-1".to_string()))
            .notification_sender(tx)
            .build();

        assert_eq!(ctx.request_id(), &RequestId::String("req-1".to_string()));
        assert!(ctx.progress_token().is_some());
    }

    #[test]
    fn test_can_sample_without_requester() {
        let ctx = RequestContext::new(RequestId::Number(1));
        assert!(!ctx.can_sample());
    }

    #[test]
    fn test_can_sample_with_requester() {
        let (request_tx, _rx) = outgoing_request_channel(10);
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(request_tx));

        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);
        assert!(ctx.can_sample());
    }

    #[tokio::test]
    async fn test_sample_without_requester_fails() {
        use crate::protocol::{CreateMessageParams, SamplingMessage};

        let ctx = RequestContext::new(RequestId::Number(1));
        let params = CreateMessageParams::new(vec![SamplingMessage::user("test")], 100);

        let result = ctx.sample(params).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Sampling not available")
        );
    }

    #[test]
    fn test_builder_with_client_requester() {
        let (request_tx, _rx) = outgoing_request_channel(10);
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(request_tx));

        let ctx = RequestContextBuilder::new()
            .request_id(RequestId::Number(1))
            .client_requester(requester)
            .build();

        assert!(ctx.can_sample());
    }

    #[test]
    fn test_can_elicit_without_requester() {
        let ctx = RequestContext::new(RequestId::Number(1));
        assert!(!ctx.can_elicit());
    }

    #[test]
    fn test_can_elicit_with_requester() {
        let (request_tx, _rx) = outgoing_request_channel(10);
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(request_tx));

        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);
        assert!(ctx.can_elicit());
    }

    #[tokio::test]
    async fn test_elicit_form_without_requester_fails() {
        use crate::protocol::{ElicitFormSchema, ElicitMode};

        let ctx = RequestContext::new(RequestId::Number(1));
        let params = ElicitFormParams {
            mode: Some(ElicitMode::Form),
            message: "Enter details".to_string(),
            requested_schema: ElicitFormSchema::new().string_field("name", None, true),
            meta: None,
        };

        let result = ctx.elicit_form(params).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Elicitation not available")
        );
    }

    #[tokio::test]
    async fn test_elicit_url_without_requester_fails() {
        use crate::protocol::ElicitMode;

        let ctx = RequestContext::new(RequestId::Number(1));
        let params = ElicitUrlParams {
            mode: Some(ElicitMode::Url),
            elicitation_id: "test-123".to_string(),
            message: "Please authorize".to_string(),
            url: "https://example.com/auth".to_string(),
            meta: None,
        };

        let result = ctx.elicit_url(params).await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Elicitation not available")
        );
    }

    #[tokio::test]
    async fn test_confirm_without_requester_fails() {
        let ctx = RequestContext::new(RequestId::Number(1));

        let result = ctx.confirm("Are you sure?").await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Elicitation not available")
        );
    }

    #[tokio::test]
    async fn test_send_log_filtered_by_level() {
        let (tx, mut rx) = notification_channel(10);
        let min_level = Arc::new(RwLock::new(LogLevel::Warning));

        let ctx = RequestContext::new(RequestId::Number(1))
            .with_notification_sender(tx)
            .with_min_log_level(min_level.clone());

        // Error is more severe than Warning — should pass through
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Error,
            serde_json::Value::Null,
        ));
        let msg = rx.try_recv();
        assert!(msg.is_ok(), "Error should pass through Warning filter");

        // Warning is equal to min level — should pass through
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Warning,
            serde_json::Value::Null,
        ));
        let msg = rx.try_recv();
        assert!(msg.is_ok(), "Warning should pass through Warning filter");

        // Info is less severe than Warning — should be filtered
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Info,
            serde_json::Value::Null,
        ));
        let msg = rx.try_recv();
        assert!(msg.is_err(), "Info should be filtered by Warning filter");

        // Debug is less severe than Warning — should be filtered
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Debug,
            serde_json::Value::Null,
        ));
        let msg = rx.try_recv();
        assert!(msg.is_err(), "Debug should be filtered by Warning filter");
    }

    #[tokio::test]
    async fn test_send_log_level_updates_dynamically() {
        let (tx, mut rx) = notification_channel(10);
        let min_level = Arc::new(RwLock::new(LogLevel::Error));

        let ctx = RequestContext::new(RequestId::Number(1))
            .with_notification_sender(tx)
            .with_min_log_level(min_level.clone());

        // Info should be filtered at Error level
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Info,
            serde_json::Value::Null,
        ));
        assert!(
            rx.try_recv().is_err(),
            "Info should be filtered at Error level"
        );

        // Dynamically update to Debug (most permissive)
        *min_level.write().unwrap() = LogLevel::Debug;

        // Now Info should pass through
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Info,
            serde_json::Value::Null,
        ));
        assert!(
            rx.try_recv().is_ok(),
            "Info should pass through after level changed to Debug"
        );
    }

    #[tokio::test]
    async fn test_send_log_no_min_level_sends_all() {
        let (tx, mut rx) = notification_channel(10);

        // No min_log_level set — all messages should pass through
        let ctx = RequestContext::new(RequestId::Number(1)).with_notification_sender(tx);

        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Debug,
            serde_json::Value::Null,
        ));
        assert!(
            rx.try_recv().is_ok(),
            "Debug should pass when no min level is set"
        );
    }

    #[tokio::test]
    #[cfg(feature = "stateless")]
    async fn final_request_log_level_is_required_and_filters_per_request() {
        let (tx, mut rx) = notification_channel(10);
        let mut extensions = Extensions::new();
        extensions.insert(crate::stateless::StatelessRequestMeta {
            protocol_version: Some(crate::protocol::PROTOCOL_VERSION_2026_07_28.to_string()),
            client_capabilities: Some(Default::default()),
            ..Default::default()
        });
        let ctx = RequestContext::new(RequestId::Number(1))
            .with_notification_sender(tx.clone())
            .with_extensions(Arc::new(extensions));
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Emergency,
            serde_json::Value::Null,
        ));
        assert!(
            rx.try_recv().is_err(),
            "final requests without logLevel must receive no logs"
        );

        let mut extensions = Extensions::new();
        extensions.insert(crate::stateless::StatelessRequestMeta {
            protocol_version: Some(crate::protocol::PROTOCOL_VERSION_2026_07_28.to_string()),
            client_capabilities: Some(Default::default()),
            log_level: Some(crate::stateless::LogLevel::Warning),
            ..Default::default()
        });
        let ctx = RequestContext::new(RequestId::Number(2))
            .with_notification_sender(tx)
            .with_extensions(Arc::new(extensions));
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Info,
            serde_json::Value::Null,
        ));
        assert!(rx.try_recv().is_err(), "Info must be filtered at Warning");
        ctx.send_log(LoggingMessageParams::new(
            LogLevel::Error,
            serde_json::Value::Null,
        ));
        assert!(rx.try_recv().is_ok(), "Error must pass at Warning");
    }

    fn make_task_object(id: &str, status: TaskStatus) -> serde_json::Value {
        serde_json::json!({
            "taskId": id,
            "status": status,
            "createdAt": "2026-04-24T00:00:00Z",
            "lastUpdatedAt": "2026-04-24T00:00:00Z",
            "ttl": null
        })
    }

    fn spawn_mock_client(
        mut rx: OutgoingRequestReceiver,
        responder: impl Fn(&str, serde_json::Value) -> serde_json::Value + Send + 'static,
    ) {
        tokio::spawn(async move {
            while let Some(req) = rx.recv().await {
                let response = responder(&req.method, req.params);
                let _ = req.response_tx.send(Ok(response));
            }
        });
    }

    #[tokio::test]
    async fn test_get_task_info_round_trips() {
        let (tx, rx) = outgoing_request_channel(10);
        spawn_mock_client(rx, |method, params| {
            assert_eq!(method, "tasks/get");
            let task_id = params["taskId"].as_str().unwrap().to_string();
            make_task_object(&task_id, TaskStatus::Working)
        });
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(tx));
        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);

        let info = ctx.get_task_info("task-123").await.unwrap();
        assert_eq!(info.task_id, "task-123");
        assert!(matches!(info.status, TaskStatus::Working));
    }

    #[tokio::test]
    #[allow(deprecated)] // exercises the legacy SEP-1686 helper
    async fn test_list_tasks_round_trips() {
        let (tx, rx) = outgoing_request_channel(10);
        spawn_mock_client(rx, |method, params| {
            assert_eq!(method, "tasks/list");
            // Status filter should be forwarded
            assert_eq!(params["status"], serde_json::json!("working"));
            serde_json::json!({
                "tasks": [
                    make_task_object("task-1", TaskStatus::Working),
                    make_task_object("task-2", TaskStatus::Working),
                ]
            })
        });
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(tx));
        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);

        let result = ctx.list_tasks(Some(TaskStatus::Working)).await.unwrap();
        assert_eq!(result.tasks.len(), 2);
        assert_eq!(result.tasks[0].task_id, "task-1");
    }

    #[tokio::test]
    async fn test_cancel_task_forwards_reason() {
        let (tx, rx) = outgoing_request_channel(10);
        spawn_mock_client(rx, |method, params| {
            assert_eq!(method, "tasks/cancel");
            assert_eq!(params["reason"], serde_json::json!("user requested"));
            // SEP-2663 (final): the cancel acknowledgment is an empty result.
            serde_json::json!({})
        });
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(tx));
        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);

        ctx.cancel_task("task-99", Some("user requested".into()))
            .await
            .expect("empty ack should succeed");
    }

    #[tokio::test]
    async fn test_cancel_task_tolerates_legacy_task_object_ack() {
        // Legacy SEP-1686 peers return the task object from tasks/cancel;
        // the helper discards the body either way.
        let (tx, rx) = outgoing_request_channel(10);
        spawn_mock_client(rx, |method, _params| {
            assert_eq!(method, "tasks/cancel");
            make_task_object("task-99", TaskStatus::Cancelled)
        });
        let requester: ClientRequesterHandle = Arc::new(ChannelClientRequester::new(tx));
        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);

        ctx.cancel_task("task-99", None)
            .await
            .expect("legacy task-object ack should also succeed");
    }

    #[tokio::test]
    async fn test_get_task_info_without_requester_fails() {
        let ctx = RequestContext::new(RequestId::Number(1));
        let result = ctx.get_task_info("task-1").await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Client request not available")
        );
    }

    #[tokio::test]
    async fn test_default_request_impl_errors() {
        // A custom requester that only implements sample/elicit (not request)
        // should reject task helpers.
        struct OnlySampleAndElicit;

        #[async_trait]
        impl ClientRequester for OnlySampleAndElicit {
            async fn sample(&self, _: CreateMessageParams) -> Result<CreateMessageResult> {
                unreachable!()
            }
            async fn elicit(&self, _: ElicitRequestParams) -> Result<ElicitResult> {
                unreachable!()
            }
        }

        let requester: ClientRequesterHandle = Arc::new(OnlySampleAndElicit);
        let ctx = RequestContext::new(RequestId::Number(1)).with_client_requester(requester);

        let err = ctx.get_task_info("x").await.unwrap_err();
        assert!(err.to_string().contains("does not support arbitrary"));
    }
}