turul-mcp-client 0.4.2

//! Main MCP client implementation

use serde_json::{Value, json};
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::time::timeout;
use tracing::{debug, info, warn};
use turul_rpc::{JsonRpcNotification, JsonRpcRequest, RequestId, RequestParams};

use crate::config::ClientConfig;
use crate::error::{McpClientError, McpClientResult, SessionError};
use crate::session::{SessionManager, SessionState};
use crate::streaming::StreamHandler;
use crate::transport::BoxedTransport;

// Re-export protocol types for convenience
use turul_mcp_protocol_2025_11_25::completion::{
    CompleteArgument, CompleteResult, CompletionContext, CompletionReference,
};
use turul_mcp_protocol_2025_11_25::meta::Cursor;
use turul_mcp_protocol_2025_11_25::resources::{ListResourceTemplatesResult, ResourceTemplate};
use turul_mcp_protocol_2025_11_25::tasks::{
    CancelTaskResult, CreateTaskResult, GetTaskResult, ListTasksResult, Task,
};
use turul_mcp_protocol_2025_11_25::{
    CallToolResult, GetPromptResult, InitializeResult, ListPromptsResult, ListResourcesResult,
    ListToolsResult, Prompt, ReadResourceResult, Resource, Tool,
};

/// Callback type for receiving server notifications.
///
/// The callback receives the notification method (e.g., `"notifications/tools/list_changed"`)
/// and the optional params object.
pub type NotificationCallback = Arc<dyn Fn(&str, Option<&Value>) + Send + Sync>;

/// Main MCP client
/// Per-tool SEP-2243 bindings: tool name → `(header name, argument JSON
/// pointer, declared type)` per `x-mcp-header` annotation.
type ParamBindingMap = std::collections::HashMap<String, Vec<(String, String, String)>>;

pub struct McpClient {
    /// Transport layer — `Arc<BoxedTransport>` (no Mutex) so concurrent
    /// `call_tool`/`list_tools`/etc. go through the transport in parallel.
    /// Each `Transport` impl is responsible for its own interior mutability.
    transport: Arc<BoxedTransport>,
    /// Session manager
    session: Arc<SessionManager>,
    /// Configuration
    config: ClientConfig,
    /// Stream handler for server events
    stream_handler: Arc<tokio::sync::Mutex<StreamHandler>>,
    /// Request ID counter
    request_counter: Arc<std::sync::atomic::AtomicU64>,
    /// Handle for the response consumer task (sends JSON-RPC responses back to server)
    response_consumer_handle: Arc<parking_lot::Mutex<Option<tokio::task::JoinHandle<()>>>>,
    /// Cached tool list (invalidated by `notifications/tools/list_changed`)
    cached_tools: Arc<RwLock<Option<Vec<Tool>>>>,
    /// SEP-2243 `Mcp-Param-*` bindings per tool, from the raw 2026 `tools/list`
    /// result. The public `Tool` vocabulary (2025 types) cannot carry the
    /// `x-mcp-header` annotation, so the bindings are captured before the
    /// type remap.
    cached_param_bindings: Arc<RwLock<ParamBindingMap>>,
    /// Raw 2026 `inputSchema` per tool, from the raw `tools/list` result. The
    /// public `Tool` vocabulary cannot hold every JSON Schema 2020-12 construct
    /// a 2026 server may advertise, so the untruncated schema is kept here.
    cached_tool_schemas: Arc<RwLock<std::collections::HashMap<String, Value>>>,
    /// Cached resource list (invalidated by `notifications/resources/list_changed`)
    cached_resources: Arc<RwLock<Option<Vec<Resource>>>>,
    /// Cached prompt list (invalidated by `notifications/prompts/list_changed`)
    cached_prompts: Arc<RwLock<Option<Vec<Prompt>>>>,
    /// The `server/discover` result captured during 2026 negotiation —
    /// capabilities, instructions, serverInfo, supportedVersions.
    discovered: Arc<RwLock<Option<DiscoveredServer>>>,
    /// User-supplied notification callback
    notification_callback: Option<NotificationCallback>,
    /// Wire spec negotiated at `connect()`, locked for this client's lifetime.
    protocol_version: Arc<RwLock<Option<crate::version::McpVersion>>>,
}

/// The body of a successful `server/discover` response, retained for the
/// lifetime of the connection. Field shapes are kept as raw JSON — the
/// public client vocabulary is version-neutral.
#[derive(Debug, Clone)]
pub struct DiscoveredServer {
    /// `serverInfo` (name/version/title…), read from the result's `_meta`.
    /// Self-reported and unverified — never key behavior or trust on it.
    pub server_info: Option<Value>,
    /// The server's declared capabilities object.
    pub capabilities: Option<Value>,
    /// Optional usage instructions ("can be used by clients to improve an
    /// LLM's understanding of available tools").
    pub instructions: Option<String>,
    /// Protocol versions the server supports.
    pub supported_versions: Vec<String>,
}

impl DiscoveredServer {
    /// Only the 2026-07-28 lane produces a `server/discover` result to parse;
    /// the narrowed 2025-11-25 build does not link the crate this reads from.
    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    fn from_result(result: &Value) -> Self {
        Self {
            server_info: result
                .get("_meta")
                .and_then(|m| m.get(turul_mcp_protocol_2026_07_28::meta::META_KEY_SERVER_INFO))
                .cloned(),
            capabilities: result.get("capabilities").cloned(),
            instructions: result
                .get("instructions")
                .and_then(|v| v.as_str())
                .map(String::from),
            supported_versions: result
                .get("supportedVersions")
                .and_then(|v| v.as_array())
                .map(|a| {
                    a.iter()
                        .filter_map(|v| v.as_str().map(String::from))
                        .collect()
                })
                .unwrap_or_default(),
        }
    }
}

/// `error.data.supported` from an UnsupportedProtocolVersionError object.
fn extract_supported_versions(error: &Value) -> Option<Vec<String>> {
    error
        .pointer("/data/supported")
        .and_then(|v| v.as_array())
        .map(|a| {
            a.iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect()
        })
}

impl Drop for McpClient {
    /// Automatic cleanup when client is dropped
    ///
    /// This ensures that if the client is dropped without explicit disconnect,
    /// we still attempt to send a DELETE request to clean up the session on the server.
    fn drop(&mut self) {
        // Abort response consumer task
        if let Some(handle) = self.response_consumer_handle.lock().take() {
            handle.abort();
        }

        let session_id = self.session.clone();
        let transport = self.transport.clone();

        // Spawn a background task to handle cleanup
        // We can't await in Drop, so we spawn a task that will complete cleanup
        tokio::spawn(async move {
            // Only send DELETE if we have a session ID
            if let Some(session_id_str) = session_id.session_id_optional().await {
                info!(
                    session_id = session_id_str,
                    "McpClient dropped - attempting session cleanup via DELETE request"
                );

                if let Err(e) = transport.send_delete(&session_id_str).await {
                    warn!(
                        session_id = session_id_str,
                        error = %e,
                        "Failed to send DELETE request during Drop cleanup"
                    );
                } else {
                    info!(
                        session_id = session_id_str,
                        "Successfully sent DELETE request during Drop cleanup"
                    );
                }
            } else {
                debug!("No session ID available, skipping DELETE request during Drop");
            }

            // Also terminate the session locally
            session_id
                .terminate(Some("Client dropped".to_string()))
                .await;
        });
    }
}

impl McpClient {
    /// Create a new MCP client with the given transport
    pub fn new(transport: BoxedTransport, config: ClientConfig) -> Self {
        Self::new_with_callback(transport, config, None)
    }

    /// Create a new MCP client with an optional notification callback
    fn new_with_callback(
        transport: BoxedTransport,
        config: ClientConfig,
        notification_callback: Option<NotificationCallback>,
    ) -> Self {
        let session = Arc::new(SessionManager::new(config.clone()));

        Self {
            transport: Arc::new(transport),
            session,
            config,
            stream_handler: Arc::new(tokio::sync::Mutex::new(StreamHandler::new())),
            request_counter: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            response_consumer_handle: Arc::new(parking_lot::Mutex::new(None)),
            cached_tools: Arc::new(RwLock::new(None)),
            cached_param_bindings: Arc::new(RwLock::new(std::collections::HashMap::new())),
            cached_tool_schemas: Arc::new(RwLock::new(std::collections::HashMap::new())),
            cached_resources: Arc::new(RwLock::new(None)),
            cached_prompts: Arc::new(RwLock::new(None)),
            discovered: Arc::new(RwLock::new(None)),
            notification_callback,
            protocol_version: Arc::new(RwLock::new(None)),
        }
    }

    /// Connect to the MCP server
    pub async fn connect(&self) -> McpClientResult<()> {
        info!("Connecting to MCP server");

        // Abort any existing response consumer before reconnecting
        if let Some(handle) = self.response_consumer_handle.lock().take() {
            handle.abort();
        }

        self.transport.connect().await?;

        // Negotiate the wire spec and run the version-appropriate handshake.
        // The listener is started AFTER this: on 2026-07-28 there is no GET SSE
        // stream to open at all, and on 2025-11-25 the GET must carry the
        // session id the handshake produces.
        self.negotiate_protocol().await?;

        self.start_server_event_listener().await?;

        info!("Successfully connected to MCP server");
        Ok(())
    }

    /// Open the transport's server-initiated event stream and wire it to the
    /// stream handler. No-op when the transport has no such stream, and skipped
    /// on 2026-07-28 — that revision removed the GET SSE stream from Streamable
    /// HTTP, so opening one only earns an HTTP 405.
    async fn start_server_event_listener(&self) -> McpClientResult<()> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            debug!("Skipping the GET SSE listener: 2026-07-28 has no GET stream");
            return Ok(());
        }

        {
            // Start event listener if supported
            if self.transport.capabilities().server_events {
                let receiver = self.transport.start_event_listener().await?;

                // Create response channel for sending JSON-RPC responses back
                let (response_tx, mut response_rx) =
                    tokio::sync::mpsc::unbounded_channel::<serde_json::Value>();

                let mut stream_handler = self.stream_handler.lock().await;
                stream_handler.set_receiver(receiver);
                stream_handler.set_response_sender(response_tx);

                // Register internal notification handler for list_changed signals
                {
                    let cached_tools = Arc::clone(&self.cached_tools);
                    let cached_resources = Arc::clone(&self.cached_resources);
                    let cached_prompts = Arc::clone(&self.cached_prompts);
                    let user_callback = self.notification_callback.clone();

                    stream_handler.on_notification(move |notification| {
                        let method = notification
                            .get("method")
                            .and_then(|m| m.as_str())
                            .unwrap_or("");
                        let params = notification.get("params");

                        match method {
                            "notifications/tools/list_changed" => {
                                info!("Server sent notifications/tools/list_changed — invalidating tool cache");
                                // Invalidate synchronously using try_write to avoid async in sync callback
                                if let Ok(mut cache) = cached_tools.try_write() {
                                    *cache = None;
                                } else {
                                    warn!("Could not acquire tool cache write lock for invalidation");
                                }
                            }
                            "notifications/resources/list_changed" => {
                                info!("Server sent notifications/resources/list_changed — invalidating resource cache");
                                if let Ok(mut cache) = cached_resources.try_write() {
                                    *cache = None;
                                } else {
                                    warn!("Could not acquire resource cache write lock for invalidation");
                                }
                            }
                            "notifications/prompts/list_changed" => {
                                info!("Server sent notifications/prompts/list_changed — invalidating prompt cache");
                                if let Ok(mut cache) = cached_prompts.try_write() {
                                    *cache = None;
                                } else {
                                    warn!("Could not acquire prompt cache write lock for invalidation");
                                }
                            }
                            _ => {
                                debug!(method = method, "Received server notification");
                            }
                        }

                        // Forward to user callback if registered
                        if let Some(ref cb) = user_callback {
                            cb(method, params);
                        }
                    });
                }

                stream_handler.start().await?;

                // Spawn consumer task that drains the channel and sends responses
                let transport_clone = Arc::clone(&self.transport);
                let consumer_handle = tokio::spawn(async move {
                    while let Some(response) = response_rx.recv().await {
                        if let Err(e) = transport_clone.send_notification(response).await {
                            warn!("Failed to send response back to server: {}", e);
                        }
                    }
                    debug!("Response consumer task stopped");
                });

                *self.response_consumer_handle.lock() = Some(consumer_handle);
            }
        }

        Ok(())
    }

    /// The wire spec negotiated for this connection, or `None` before `connect()`.
    pub async fn negotiated_version(&self) -> Option<crate::version::McpVersion> {
        *self.protocol_version.read().await
    }

    async fn lock_version(&self, v: crate::version::McpVersion) -> McpClientResult<()> {
        *self.protocol_version.write().await = Some(v);
        // Advertise the negotiated spec on the MCP-Protocol-Version header (and, for
        // the 2026-07-28 stateless core, stop sending the removed Mcp-Session-Id).
        self.transport.set_protocol_version(v.as_str());
        // 2026-07-28 is stateless — there is no initialize handshake to mark the
        // session ready, so do it here. (A 2025-11-25 connection is already Active
        // via initialize_session() before this is reached.)
        if v == crate::version::McpVersion::V2026_07_28 {
            self.session
                .set_state(crate::session::SessionState::Active)
                .await;
        }
        info!(version = %v, "Negotiated MCP wire version");
        Ok(())
    }

    /// Bilingual negotiation: explicit hint wins; otherwise probe `server/discover`
    /// and apply the [`classify_probe`](crate::version::classify_probe) rule.
    #[cfg(feature = "client-bilingual")]
    async fn negotiate_protocol(&self) -> McpClientResult<()> {
        use crate::version::{McpVersion, ProbeDecision, classify_probe};

        if let Some(hint) = self.config.mcp_protocol_version {
            match hint {
                McpVersion::V2025_11_25 => self.initialize_session().await?,
                // 2026-07-28 stateless core: no initialize handshake.
                McpVersion::V2026_07_28 => {}
            }
            return self.lock_version(hint).await;
        }

        let probe = self.probe_discover().await?;
        match classify_probe(probe, self.config.allow_legacy_gateway_fallback) {
            ProbeDecision::Use2026 => self.lock_version(McpVersion::V2026_07_28).await,
            ProbeDecision::FallbackTo2025 => {
                // The probe advertised 2026-07-28 on the wire; restore the legacy
                // version header before the initialize handshake.
                self.transport
                    .set_protocol_version(McpVersion::V2025_11_25.as_str());
                self.initialize_session().await?;
                self.lock_version(McpVersion::V2025_11_25).await
            }
            ProbeDecision::Abort(reason) => {
                Err(crate::error::ProtocolError::NegotiationFailed(reason).into())
            }
        }
    }

    #[cfg(feature = "client-2025-11-25-only")]
    async fn negotiate_protocol(&self) -> McpClientResult<()> {
        self.initialize_session().await?;
        self.lock_version(crate::version::McpVersion::V2025_11_25)
            .await
    }

    #[cfg(feature = "client-2026-07-28-only")]
    async fn negotiate_protocol(&self) -> McpClientResult<()> {
        match self.probe_discover().await? {
            crate::version::DiscoverProbe::Discovered => {
                self.lock_version(crate::version::McpVersion::V2026_07_28).await
            }
            _ => Err(crate::error::ProtocolError::UnsupportedVersion(
                "client-2026-07-28-only: server did not answer server/discover — not a 2026-07-28 server"
                    .to_string(),
            )
            .into()),
        }
    }

    /// Probe the server with a `server/discover` request and classify the outcome
    /// into a [`DiscoverProbe`](crate::version::DiscoverProbe). A valid result =>
    /// 2026; a JSON-RPC error => carries the code; an HTTP non-2xx => carries the status.
    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    async fn probe_discover(&self) -> McpClientResult<crate::version::DiscoverProbe> {
        use crate::error::TransportError;
        use crate::version::DiscoverProbe;
        use turul_mcp_protocol_2026_07_28 as p2026;

        let meta = p2026::meta::RequestMetaObject::new(
            p2026::MCP_VERSION,
            p2026::initialize::Implementation::new(
                self.config.client_info.name.clone(),
                self.config.client_info.version.clone(),
            ),
            p2026::initialize::ClientCapabilities::default(),
        );
        let discover = p2026::discover::DiscoverRequest::new(meta);
        let params = serde_json::to_value(&discover.params).map_err(|e| {
            McpClientError::generic(format!("Failed to serialize server/discover params: {}", e))
        })?;
        let envelope = self.build_request("server/discover", params);

        // The probe is a 2026-era request: its `_meta` says 2026-07-28, and the
        // MCP-Protocol-Version header MUST match the body (a 2026 server rejects
        // the disagreement with 400). Advertise 2026-07-28 for the probe; the
        // FallbackTo2025 arm restores the legacy header before `initialize`.
        self.transport.set_protocol_version(p2026::MCP_VERSION);

        match timeout(
            self.config.timeouts.initialization,
            self.send_request_with_headers_internal(envelope),
        )
        .await
        {
            Err(_) => Err(McpClientError::Timeout),
            Ok(Ok(resp)) => {
                if let Some(err) = resp.body.get("error") {
                    let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
                    Ok(DiscoverProbe::JsonRpcError(
                        code,
                        extract_supported_versions(err),
                    ))
                } else if let Some(result) = resp.body.get("result") {
                    // Retain the DiscoverResult body — capabilities,
                    // instructions, serverInfo, supportedVersions — for the
                    // accessors ("Both parties must respect declared
                    // capabilities throughout the interaction").
                    *self.discovered.write().await = Some(DiscoveredServer::from_result(result));
                    Ok(DiscoverProbe::Discovered)
                } else {
                    Err(crate::error::ProtocolError::InvalidResponse(
                        "server/discover response carried neither result nor error".to_string(),
                    )
                    .into())
                }
            }
            Ok(Err(McpClientError::Transport(TransportError::HttpStatus { status, message }))) => {
                // A 400 whose body is a recognized modern JSON-RPC error is a
                // negotiation signal, not a transport failure — surface the code
                // (e.g. -32022 UnsupportedProtocolVersionError) to the classifier.
                if let Ok(body) = serde_json::from_str::<serde_json::Value>(&message)
                    && let Some(code) = body.pointer("/error/code").and_then(|c| c.as_i64())
                {
                    let supported = body.get("error").and_then(extract_supported_versions);
                    return Ok(DiscoverProbe::JsonRpcError(code, supported));
                }
                Ok(DiscoverProbe::HttpStatus(status))
            }
            Ok(Err(other)) => Err(other),
        }
    }

    /// Disconnect from the MCP server
    pub async fn disconnect(&self) -> McpClientResult<()> {
        info!("Disconnecting from MCP server");

        // Stop response consumer task
        if let Some(handle) = self.response_consumer_handle.lock().take() {
            handle.abort();
        }

        // Send DELETE request for session cleanup if we have a session ID
        if let Some(session_id) = self.session.session_id_optional().await {
            if let Err(e) = self.transport.send_delete(&session_id).await {
                warn!(
                    session_id = session_id,
                    error = %e,
                    "Failed to send DELETE request during disconnect - continuing with cleanup"
                );
            }
        } else {
            debug!("No session ID available, skipping DELETE request during disconnect");
        }

        // Terminate session locally
        self.session
            .terminate(Some("Client disconnect".to_string()))
            .await;

        // Disconnect transport
        self.transport.disconnect().await?;

        info!("Disconnected from MCP server");
        Ok(())
    }

    /// Update (or clear) the bearer token sent on subsequent transport requests.
    ///
    /// This is the canonical way to rotate an OAuth `client_credentials` token
    /// on a long-lived [`McpClient`]. Callers should invoke this *before* the
    /// next outbound request (most importantly, before [`Self::disconnect`]),
    /// so the resulting DELETE flies under the fresh bearer rather than the
    /// one that was baked into the transport at construction.
    ///
    /// Passing `None` removes the override and falls back to whatever
    /// `Authorization` header (if any) was configured at construction.
    ///
    /// On transports that don't authenticate at the transport layer (stdio,
    /// plain SSE), this is a no-op — see [`Transport::update_auth_header`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Old client is about to be torn down after the M2M token rotated.
    /// old_client.set_bearer(Some(&fresh_token)).await;
    /// old_client.disconnect().await?;   // DELETE under fresh bearer
    /// ```
    pub async fn set_bearer(&self, token: Option<&str>) {
        let value = token.map(|t| format!("Bearer {}", t));
        self.transport.update_auth_header(value).await;
    }

    /// Check if client is connected and ready
    pub async fn is_ready(&self) -> bool {
        let transport_connected = self.transport.is_connected();
        let session_ready = self.session.is_ready().await;

        transport_connected && session_ready
    }

    /// Get client connection status
    pub async fn connection_status(&self) -> ConnectionStatus {
        let transport_info = self.transport.connection_info();

        let session_stats = self.session.statistics().await;

        ConnectionStatus {
            transport_connected: transport_info.connected,
            session_state: session_stats.state,
            transport_type: transport_info.transport_type,
            endpoint: transport_info.endpoint,
            session_id: session_stats.session_id,
            protocol_version: session_stats.protocol_version,
        }
    }

    /// Initialize session with server
    async fn initialize_session(&self) -> McpClientResult<()> {
        info!("Initializing MCP session");

        self.session.mark_initializing().await?;

        let init_request = self.session.create_initialize_request().await;
        let request_json = serde_json::to_value(&init_request).map_err(|e| {
            McpClientError::generic(format!("Failed to serialize initialize request: {}", e))
        })?;

        let json_rpc_request = self.build_request("initialize", request_json);

        // Send initialize request with timeout (need headers for session ID)
        let response = timeout(
            self.config.timeouts.initialization,
            self.send_request_with_headers_internal(json_rpc_request),
        )
        .await
        .map_err(|_| McpClientError::Timeout)?;

        let transport_response = response?;

        // Extract session ID from headers (MCP protocol) - case insensitive lookup
        let session_id = transport_response
            .headers
            .iter()
            .find(|(key, _)| key.to_lowercase() == "mcp-session-id")
            .map(|(_, value)| value.clone());

        if let Some(session_id) = session_id {
            info!("Server provided session ID: {}", session_id);

            // Store in session manager
            self.session.set_session_id(session_id.clone()).await?;

            // Tell transport to include session ID in all subsequent requests
            self.transport.set_session_id(session_id);
        } else {
            debug!("Server did not provide Mcp-Session-Id — stateless session (spec-valid)");
        }

        // Parse initialize response
        let init_response: InitializeResult = serde_json::from_value(
            transport_response
                .body
                .get("result")
                .cloned()
                .unwrap_or(Value::Null),
        )
        .map_err(|e| {
            McpClientError::generic(format!("Failed to parse initialize response: {}", e))
        })?;

        // Validate negotiated protocol version
        SessionManager::validate_protocol_version(&init_response.protocol_version)?;

        // Validate server capabilities
        self.session
            .validate_server_capabilities(&init_response.capabilities)
            .await?;

        // Complete session initialization
        self.session
            .initialize(
                init_request.capabilities,
                init_response.capabilities,
                init_response.protocol_version,
            )
            .await?;

        // Send initialized notification per MCP spec
        let initialized_notification =
            Self::build_notification("notifications/initialized", json!({}));

        self.send_notification_internal(initialized_notification)
            .await?;

        info!("MCP session initialized successfully");
        Ok(())
    }

    /// Generate next request ID
    fn next_request_id(&self) -> String {
        let counter = self
            .request_counter
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        format!("req_{}", counter)
    }

    /// Build a JSON-RPC request envelope as a `Value` ready for the transport.
    ///
    /// Routes through `turul_rpc::JsonRpcRequest` so the `jsonrpc: "2.0"` envelope
    /// is single-source-of-truth — no manual `json!({"jsonrpc": "2.0", ...})` at
    /// call sites. An empty `Value::Object` is preserved as `"params":{}` on the
    /// wire (not omitted) for drop-in compatibility with the prior hand-rolled form.
    fn build_request(&self, method: &str, params: Value) -> Value {
        let id = RequestId::String(self.next_request_id());
        let request = JsonRpcRequest::new(id, method.to_string(), value_to_request_params(params));
        serde_json::to_value(request).expect("JsonRpcRequest serialization is infallible")
    }

    /// Build a JSON-RPC notification envelope as a `Value` ready for the transport.
    fn build_notification(method: &str, params: Value) -> Value {
        let notification =
            JsonRpcNotification::new(method.to_string(), value_to_request_params(params));
        serde_json::to_value(notification).expect("JsonRpcNotification serialization is infallible")
    }

    /// Send request and handle retries
    async fn send_request_internal(&self, request: Value) -> McpClientResult<Value> {
        let mut last_error = None;

        for attempt in 0..self.config.retry.max_attempts {
            if attempt > 0 {
                let delay = self.config.retry.delay_for_attempt(attempt);
                debug!(
                    attempt = attempt,
                    delay_ms = delay.as_millis(),
                    "Retrying request"
                );
                tokio::time::sleep(delay).await;
            }

            match self.send_request_raw(request.clone()).await {
                Ok(response) => {
                    self.session.update_activity().await;
                    return Ok(response);
                }
                Err(e) => {
                    warn!(attempt = attempt, error = %e, "Request failed");

                    // 404 means "session unknown, re-initialize" only where
                    // sessions exist. On 2026-07-28 it is the mapping for an
                    // unknown method, so recovering here would answer a method
                    // the peer does not implement by sending `initialize` — a
                    // method that revision removed. Surface the 404 instead.
                    if e.is_session_expired()
                        && self.negotiated_version().await
                            != Some(crate::version::McpVersion::V2026_07_28)
                    {
                        warn!("Session expired (HTTP 404) — attempting re-initialization");
                        self.session.reset().await;
                        // Clear stale session ID from transport so initialize
                        // request is sent without Mcp-Session-Id header
                        self.transport.clear_session_id();
                        if let Err(reinit_err) = self.initialize_session().await {
                            warn!(error = %reinit_err, "Re-initialization failed");
                            return Err(e);
                        }
                        // Retry the request with the new session
                        continue;
                    }

                    // JSON-RPC -32031: server rejected because notifications/initialized
                    // hasn't been processed yet. Disconnect, full reconnect, retry once.
                    if e.is_session_not_initialized() {
                        warn!(
                            "Session not initialized (code -32031) — \
                             disconnecting and reconnecting"
                        );
                        // Full reconnect: disconnect old session, connect fresh
                        if let Err(dc_err) = self.disconnect().await {
                            warn!(error = %dc_err, "Disconnect during session retry failed");
                        }
                        self.session.reset().await;
                        if let Err(reconnect_err) = self.connect().await {
                            warn!(error = %reconnect_err, "Reconnect after -32031 failed");
                            return Err(e);
                        }
                        // Retry the original request exactly once
                        return match self.send_request_raw(request).await {
                            Ok(response) => {
                                self.session.update_activity().await;
                                Ok(response)
                            }
                            Err(retry_err) => {
                                warn!(error = %retry_err, "Retry after reconnect also failed");
                                Err(retry_err)
                            }
                        };
                    }

                    if !e.is_retryable() || !self.config.retry.should_retry(attempt + 1) {
                        return Err(e);
                    }

                    last_error = Some(e);
                }
            }
        }

        Err(last_error.unwrap_or_else(|| McpClientError::generic("All retry attempts failed")))
    }

    /// Send request with headers and handle retries (used for initialization)
    async fn send_request_with_headers_internal(
        &self,
        request: Value,
    ) -> McpClientResult<crate::transport::TransportResponse> {
        let mut last_error = None;

        for attempt in 0..self.config.retry.max_attempts {
            if attempt > 0 {
                let delay = self.config.retry.delay_for_attempt(attempt);
                debug!(
                    attempt = attempt,
                    delay_ms = delay.as_millis(),
                    "Retrying request with headers"
                );
                tokio::time::sleep(delay).await;
            }

            match self.send_request_with_headers_raw(request.clone()).await {
                Ok(response) => return Ok(response),
                Err(e) => {
                    warn!(
                        attempt = attempt,
                        max_attempts = self.config.retry.max_attempts,
                        error = %e,
                        "Request with headers failed"
                    );

                    if !e.is_retryable() {
                        return Err(e);
                    }

                    last_error = Some(e);
                }
            }
        }

        Err(last_error.unwrap_or_else(|| McpClientError::generic("All retry attempts failed")))
    }

    /// Send raw request with headers without retries
    async fn send_request_with_headers_raw(
        &self,
        request: Value,
    ) -> McpClientResult<crate::transport::TransportResponse> {
        timeout(
            self.config.timeouts.request,
            self.transport.send_request_with_headers(request),
        )
        .await
        .map_err(|_| McpClientError::Timeout)?
    }

    /// Send raw request without retries
    async fn send_request_raw(&self, request: Value) -> McpClientResult<Value> {
        if !self.session.is_ready().await {
            return Err(SessionError::NotInitialized.into());
        }

        let response = timeout(
            self.config.timeouts.request,
            self.transport.send_request(request),
        )
        .await
        .map_err(|_| McpClientError::Timeout)??;

        // Check for JSON-RPC error
        if let Some(error) = response.get("error") {
            let code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(-1) as i32;
            let message = error
                .get("message")
                .and_then(|m| m.as_str())
                .unwrap_or("Unknown error");
            let data = error.get("data").cloned();

            return Err(McpClientError::server_error(code, message, data));
        }

        Ok(response)
    }

    /// Send notification
    async fn send_notification_internal(&self, notification: Value) -> McpClientResult<()> {
        self.transport.send_notification(notification).await?;
        self.session.update_activity().await;
        Ok(())
    }

    /// List available tools (returns cached result if available)
    ///
    /// Returns the FIRST page only — pagination spec: "Treat a missing
    /// nextCursor as the end of results". Use
    /// [`list_tools_paginated`](Self::list_tools_paginated) to walk all pages.
    ///
    /// The cache is automatically invalidated when the server sends a
    /// `notifications/tools/list_changed` notification. Use [`refresh_tools`](Self::refresh_tools)
    /// to force a fresh fetch.
    ///
    /// # Schema fidelity on a 2026-07-28 connection
    ///
    /// The returned [`Tool`] models each `inputSchema` property as a closed enum
    /// keyed on `"type"`, so a property expressed with a JSON Schema 2020-12
    /// composition keyword (`oneOf`/`anyOf`/`allOf`/`$ref` — legal per SEP-2106)
    /// cannot be carried. Such a property is dropped from the returned schema and
    /// a `tracing::warn!` naming the tool and the dropped paths is emitted; the
    /// tool itself is still listed and still callable. Read
    /// [`tool_input_schema`](Self::tool_input_schema) for the untruncated schema
    /// the server actually advertised.
    pub async fn list_tools(&self) -> McpClientResult<Vec<Tool>> {
        // Return cached tools if available
        {
            let cache = self.cached_tools.read().await;
            if let Some(ref tools) = *cache {
                debug!(count = tools.len(), "Returning cached tools");
                return Ok(tools.clone());
            }
        }

        let tools = self.fetch_tools().await?;

        // Update cache
        {
            let mut cache = self.cached_tools.write().await;
            *cache = Some(tools.clone());
        }

        Ok(tools)
    }

    /// The untruncated `inputSchema` the server advertised for `tool_name`, as
    /// raw JSON — populated by [`list_tools`](Self::list_tools) /
    /// [`refresh_tools`](Self::refresh_tools) on a 2026-07-28 connection.
    ///
    /// This is the full-fidelity route for the JSON Schema 2020-12 constructs
    /// the public [`Tool`] type cannot hold (see
    /// [`list_tools`](Self::list_tools)). `None` on a 2025-11-25 connection, or
    /// before any tool listing has been fetched.
    pub async fn tool_input_schema(&self, tool_name: &str) -> Option<Value> {
        self.cached_tool_schemas
            .read()
            .await
            .get(tool_name)
            .cloned()
    }

    /// Force a fresh fetch of tools from the server, bypassing and updating the cache.
    ///
    /// Same 2026-07-28 schema-fidelity caveat as [`list_tools`](Self::list_tools).
    pub async fn refresh_tools(&self) -> McpClientResult<Vec<Tool>> {
        let tools = self.fetch_tools().await?;

        // Update cache
        {
            let mut cache = self.cached_tools.write().await;
            *cache = Some(tools.clone());
        }

        Ok(tools)
    }

    /// Fetch tools from the server (no cache interaction).
    async fn fetch_tools(&self) -> McpClientResult<Vec<Tool>> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return self.fetch_tools_2026_07_28().await;
        }
        self.fetch_tools_2025_11_25().await
    }

    /// `tools/list` for a 2025-11-25 connection (the historical alias path).
    async fn fetch_tools_2025_11_25(&self) -> McpClientResult<Vec<Tool>> {
        debug!("Fetching tools from server (2025-11-25)");

        let request = self.build_request("tools/list", json!({}));

        let response = self.send_request_internal(request).await?;
        let tools_response: ListToolsResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(count = tools_response.tools.len(), "Retrieved tools");
        Ok(tools_response.tools)
    }

    /// `tools/list` for a 2026-07-28 connection: request carries `_meta`, and the
    /// 2026-shaped result (`resultType`/`ttlMs`/`cacheScope` + tools) is parsed
    /// via the [`protocol::v2026`](crate::protocol::v2026_07_28) module.
    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    async fn fetch_tools_2026_07_28(&self) -> McpClientResult<Vec<Tool>> {
        debug!("Fetching tools from server (2026-07-28)");

        let meta = crate::protocol::v2026_07_28::request_meta(
            &self.config.client_info.name,
            &self.config.client_info.version,
            &self.config.declared_capabilities,
        );
        let request = self.build_request(
            "tools/list",
            crate::protocol::v2026_07_28::params_with_meta(&meta, json!({})),
        );

        let response = self.send_request_internal(request).await?;
        let raw_result = response.get("result").cloned().unwrap_or(Value::Null);
        let tools = crate::protocol::v2026_07_28::parse_list_tools(&raw_result)?;

        // Capture the SEP-2243 Mcp-Param bindings and the untruncated
        // inputSchemas before the vocabulary remap drops the x-mcp-header
        // annotations and any 2020-12 construct the public type cannot hold.
        {
            let bindings = crate::protocol::v2026_07_28::collect_param_bindings(&raw_result);
            *self.cached_param_bindings.write().await = bindings;
            let schemas = crate::protocol::v2026_07_28::collect_input_schemas(&raw_result);
            *self.cached_tool_schemas.write().await = schemas;
        }

        debug!(count = tools.len(), "Retrieved tools");
        Ok(tools)
    }

    /// List available tools with pagination support
    ///
    /// Same 2026-07-28 schema-fidelity caveat as [`list_tools`](Self::list_tools).
    pub async fn list_tools_paginated(
        &self,
        cursor: Option<Cursor>,
    ) -> McpClientResult<ListToolsResult> {
        debug!("Listing tools with pagination");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let extra = match cursor {
                Some(ref c) => json!({ "cursor": c.as_str() }),
                None => json!({}),
            };
            return self
                .send_2026_07_28(
                    "tools/list",
                    extra,
                    crate::protocol::v2026_07_28::parse_list_tools_result,
                )
                .await;
        }

        let request_params = if let Some(cursor) = cursor {
            json!({ "cursor": cursor.as_str() })
        } else {
            json!({})
        };

        let request = self.build_request("tools/list", request_params);

        let response = self.send_request_internal(request).await?;
        let tools_response: ListToolsResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = tools_response.tools.len(),
            has_cursor = tools_response.next_cursor.is_some(),
            "Retrieved tools with pagination"
        );
        Ok(tools_response)
    }

    /// Issue a 2026-07-28 operation: build the request with the required per-request
    /// `_meta`, send it, and parse the 2026-shaped result via the supplied parser.
    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    async fn send_2026_07_28<T>(
        &self,
        method: &str,
        extra: Value,
        parse: impl Fn(&Value) -> McpClientResult<T>,
    ) -> McpClientResult<T> {
        let meta = crate::protocol::v2026_07_28::request_meta(
            &self.config.client_info.name,
            &self.config.client_info.version,
            &self.config.declared_capabilities,
        );
        let request = self.build_request(
            method,
            crate::protocol::v2026_07_28::params_with_meta(&meta, extra),
        );
        let response = self.send_request_internal(request).await?;
        parse(&response.get("result").cloned().unwrap_or(Value::Null))
    }

    /// SEP-2243: the `Mcp-Param-*` headers for a `tools/call`, from the cached
    /// bindings. Empty when the tool schema has not been fetched yet — per the
    /// spec the call then goes out without custom headers, and a rejecting
    /// server prompts the application to `list_tools()` and retry.
    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    async fn mcp_param_headers_for(
        &self,
        tool_name: &str,
        arguments: &Value,
    ) -> Vec<(String, String)> {
        let cache = self.cached_param_bindings.read().await;
        match cache.get(tool_name) {
            Some(bindings) => {
                crate::protocol::v2026_07_28::encode_param_headers(bindings, arguments)
            }
            None => Vec::new(),
        }
    }

    #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
    async fn send_2026_07_28_with_extra_headers<T>(
        &self,
        method: &str,
        extra: Value,
        extra_headers: &[(String, String)],
        parse: impl Fn(&Value) -> McpClientResult<T>,
    ) -> McpClientResult<T> {
        let meta = crate::protocol::v2026_07_28::request_meta(
            &self.config.client_info.name,
            &self.config.client_info.version,
            &self.config.declared_capabilities,
        );
        let request = self.build_request(
            method,
            crate::protocol::v2026_07_28::params_with_meta(&meta, extra),
        );
        if !self.session.is_ready().await {
            return Err(crate::error::SessionError::NotInitialized.into());
        }
        let response = tokio::time::timeout(
            self.config.timeouts.request,
            self.transport
                .send_request_with_extra_headers(request, extra_headers),
        )
        .await
        .map_err(|_| McpClientError::Timeout)??;
        if let Some(error) = response.get("error") {
            let code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(-1) as i32;
            let message = error
                .get("message")
                .and_then(|m| m.as_str())
                .unwrap_or("Unknown error");
            let data = error.get("data").cloned();
            return Err(McpClientError::server_error(code, message, data));
        }
        parse(&response.get("result").cloned().unwrap_or(Value::Null))
    }

    /// Reject an operation whose method was removed from the 2026-07-28 core when the
    /// connection negotiated 2026. The method remains available on a 2025-11-25 connection.
    async fn reject_if_2026_07_28(&self, _method: &str) -> McpClientResult<()> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return Err(crate::error::ProtocolError::MethodNotFound(format!(
                "`{_method}` is not part of MCP 2026-07-28 (removed from core); this connection negotiated 2026"
            ))
            .into());
        }
        Ok(())
    }

    /// Call a tool
    pub async fn call_tool(&self, name: &str, arguments: Value) -> McpClientResult<CallToolResult> {
        debug!(tool = name, "Calling tool");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let extra_headers = self.mcp_param_headers_for(name, &arguments).await;
            let first = self
                .send_2026_07_28_with_extra_headers(
                    "tools/call",
                    json!({ "name": name, "arguments": arguments.clone() }),
                    &extra_headers,
                    crate::protocol::v2026_07_28::parse_call_tool,
                )
                .await;
            // SEP-2243 §Client Behavior: on a header-mismatch rejection "the
            // client SHOULD call tools/list to obtain the current inputSchema,
            // then retry the original request with the appropriate headers."
            // One refresh + one retry; a second mismatch error surfaces to
            // the caller.
            let is_header_mismatch = matches!(
                &first,
                Err(McpClientError::ServerError { code, .. })
                    if *code == turul_mcp_protocol_2026_07_28::headers::ERROR_CODE_HEADER_MISMATCH as i32
            );
            if is_header_mismatch {
                debug!(
                    tool = name,
                    "Mcp-Param mismatch — refreshing tools/list and retrying once"
                );
                let _ = self.refresh_tools().await?;
                let extra_headers = self.mcp_param_headers_for(name, &arguments).await;
                return self
                    .send_2026_07_28_with_extra_headers(
                        "tools/call",
                        json!({ "name": name, "arguments": arguments.clone() }),
                        &extra_headers,
                        crate::protocol::v2026_07_28::parse_call_tool,
                    )
                    .await;
            }
            return first;
        }

        let request = self.build_request(
            "tools/call",
            json!({
                "name": name,
                "arguments": arguments
            }),
        );

        let response = self.send_request_internal(request).await?;
        let call_response: CallToolResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            tool = name,
            is_error = call_response.is_error,
            "Tool call completed"
        );
        Ok(call_response)
    }

    /// Open a `subscriptions/listen` stream (2026-07-28 connections only).
    ///
    /// `filter` is the wire `SubscriptionFilter` object (e.g.
    /// `{"toolsListChanged": true, "resourceSubscriptions": ["file:///a"]}`).
    /// Returns the stream AFTER consuming and validating the mandatory first
    /// message (`notifications/subscriptions/acknowledged`); the returned
    /// [`SubscriptionStream`] exposes the honored filter and the subscription
    /// id, then yields each subsequent notification. Dropping the stream
    /// closes it, which per Streamable HTTP cancels the subscription.
    pub async fn subscriptions_listen(&self, filter: Value) -> McpClientResult<SubscriptionStream> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let meta = crate::protocol::v2026_07_28::request_meta(
                &self.config.client_info.name,
                &self.config.client_info.version,
                &self.config.declared_capabilities,
            );
            let request = self.build_request(
                "subscriptions/listen",
                crate::protocol::v2026_07_28::params_with_meta(
                    &meta,
                    json!({ "notifications": filter }),
                ),
            );
            let request_id = request
                .get("id")
                .and_then(|id| id.as_str())
                .map(String::from);
            let mut receiver = self.transport.send_request_streaming(request).await?;

            // The server MUST send the acknowledgement first.
            let ack = tokio::time::timeout(self.config.timeouts.request, receiver.recv())
                .await
                .map_err(|_| McpClientError::Timeout)?
                .ok_or_else(|| {
                    crate::error::ProtocolError::InvalidResponse(
                        "subscriptions/listen stream closed before the acknowledgement".to_string(),
                    )
                })?;
            if ack.get("method").and_then(|m| m.as_str())
                != Some("notifications/subscriptions/acknowledged")
            {
                return Err(crate::error::ProtocolError::InvalidResponse(format!(
                    "first listen-stream message must be the acknowledgement, got: {ack}"
                ))
                .into());
            }
            let honored = ack
                .pointer("/params/notifications")
                .cloned()
                .unwrap_or(Value::Null);
            let subscription_id = ack
                .pointer("/params/_meta/io.modelcontextprotocol~1subscriptionId")
                .or_else(|| {
                    ack.get("params")
                        .and_then(|p| p.get("_meta"))
                        .and_then(|m| m.get("io.modelcontextprotocol/subscriptionId"))
                })
                .and_then(|v| v.as_str())
                .map(String::from);
            return Ok(SubscriptionStream {
                honored,
                subscription_id,
                request_id,
                receiver,
            });
        }

        let _ = filter;
        Err(crate::error::ProtocolError::MethodNotFound(
            "subscriptions/listen requires a 2026-07-28 connection".to_string(),
        )
        .into())
    }

    /// The `server/discover` result captured at connect time (2026
    /// connections): server capabilities, instructions, supported versions.
    /// `None` before connect or on a 2025-11-25 connection.
    pub async fn discovered_server(&self) -> Option<DiscoveredServer> {
        self.discovered.read().await.clone()
    }

    /// The server's declared capabilities from `server/discover` —
    /// "Both parties must respect declared capabilities throughout the
    /// interaction".
    pub async fn server_capabilities(&self) -> Option<Value> {
        self.discovered
            .read()
            .await
            .as_ref()
            .and_then(|d| d.capabilities.clone())
    }

    /// The server's `instructions` from `server/discover`, for priming an
    /// LLM with usage guidance.
    pub async fn server_instructions(&self) -> Option<String> {
        self.discovered
            .read()
            .await
            .as_ref()
            .and_then(|d| d.instructions.clone())
    }

    /// Call a tool with a request-scoped progress feed (2026-07-28
    /// connections only). The `progress_token` (string or number) rides
    /// `_meta.progressToken`; each `notifications/progress` arriving on the
    /// request's SSE stream is handed to `on_progress` before the final
    /// result is returned.
    pub async fn call_tool_with_progress(
        &self,
        name: &str,
        arguments: Value,
        progress_token: Value,
        mut on_progress: impl FnMut(Value) + Send,
    ) -> McpClientResult<CallToolResult> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let meta = crate::protocol::v2026_07_28::request_meta(
                &self.config.client_info.name,
                &self.config.client_info.version,
                &self.config.declared_capabilities,
            );
            let mut params = crate::protocol::v2026_07_28::params_with_meta(
                &meta,
                json!({ "name": name, "arguments": arguments }),
            );
            if let Some(m) = params.get_mut("_meta").and_then(|m| m.as_object_mut()) {
                m.insert("progressToken".to_string(), progress_token);
            }
            let request = self.build_request("tools/call", params);
            let mut receiver = self.transport.send_request_streaming(request).await?;

            let deadline = tokio::time::Instant::now() + self.config.timeouts.request;
            loop {
                let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
                let frame = tokio::time::timeout(remaining, receiver.recv())
                    .await
                    .map_err(|_| McpClientError::Timeout)?
                    .ok_or_else(|| {
                        crate::error::ProtocolError::InvalidResponse(
                            "request stream closed before the final response".to_string(),
                        )
                    })?;
                if frame.get("method").and_then(|m| m.as_str()) == Some("notifications/progress") {
                    on_progress(frame.get("params").cloned().unwrap_or(Value::Null));
                    continue;
                }
                if let Some(err) = frame.get("error") {
                    return Err(McpClientError::server_error(
                        err.get("code").and_then(|c| c.as_i64()).unwrap_or(0) as i32,
                        err.get("message")
                            .and_then(|m| m.as_str())
                            .unwrap_or("server error"),
                        err.get("data").cloned(),
                    ));
                }
                if frame.get("result").is_some() {
                    return crate::protocol::v2026_07_28::parse_call_tool(
                        frame.get("result").unwrap_or(&Value::Null),
                    );
                }
                // Other request-scoped notifications (e.g. notifications/message)
                // are not progress — skip.
            }
        }

        let _ = (name, arguments, progress_token, &mut on_progress);
        Err(crate::error::ProtocolError::MethodNotFound(
            "per-request progress requires a 2026-07-28 connection".to_string(),
        )
        .into())
    }

    /// Retry a `tools/call` after an `InputRequired` outcome (MRTR, SEP-2322):
    /// re-issues the ORIGINAL request (same name/arguments) with the gathered
    /// `input_responses` and the server's `request_state` echoed verbatim,
    /// under a fresh JSON-RPC id. 2026-07-28 connections only.
    pub async fn call_tool_with_input_responses(
        &self,
        name: &str,
        arguments: Value,
        input_responses: Value,
        request_state: Option<String>,
    ) -> McpClientResult<CallToolResult> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let mut extra = json!({
                "name": name,
                "arguments": arguments,
                "inputResponses": input_responses,
            });
            if let (Some(obj), Some(state)) = (extra.as_object_mut(), request_state) {
                obj.insert("requestState".to_string(), Value::String(state));
            }
            return self
                .send_2026_07_28(
                    "tools/call",
                    extra,
                    crate::protocol::v2026_07_28::parse_call_tool,
                )
                .await;
        }

        let _ = (name, arguments, input_responses, request_state);
        Err(crate::error::ProtocolError::MethodNotFound(
            "MRTR input responses require a 2026-07-28 connection".to_string(),
        )
        .into())
    }

    /// Retry a `resources/read` after an `InputRequired` outcome (MRTR,
    /// SEP-2322): re-issues the ORIGINAL request (same uri) with the gathered
    /// `input_responses` and the server's `request_state` echoed verbatim,
    /// under a fresh JSON-RPC id. 2026-07-28 connections only.
    pub async fn read_resource_with_input_responses(
        &self,
        uri: &str,
        input_responses: Value,
        request_state: Option<String>,
    ) -> McpClientResult<Vec<turul_mcp_protocol_2025_11_25::ResourceContent>> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let mut extra = json!({
                "uri": uri,
                "inputResponses": input_responses,
            });
            if let (Some(obj), Some(state)) = (extra.as_object_mut(), request_state) {
                obj.insert("requestState".to_string(), Value::String(state));
            }
            let r = self
                .send_2026_07_28(
                    "resources/read",
                    extra,
                    crate::protocol::v2026_07_28::parse_read_resource,
                )
                .await?;
            return Ok(r.contents);
        }

        let _ = (uri, input_responses, request_state);
        Err(crate::error::ProtocolError::MethodNotFound(
            "MRTR input responses require a 2026-07-28 connection".to_string(),
        )
        .into())
    }

    /// Retry a `prompts/get` after an `InputRequired` outcome (MRTR,
    /// SEP-2322): re-issues the ORIGINAL request (same name/arguments) with
    /// the gathered `input_responses` and the server's `request_state` echoed
    /// verbatim, under a fresh JSON-RPC id. 2026-07-28 connections only.
    pub async fn get_prompt_with_input_responses(
        &self,
        name: &str,
        arguments: Option<Value>,
        input_responses: Value,
        request_state: Option<String>,
    ) -> McpClientResult<GetPromptResult> {
        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let mut extra = json!({
                "name": name,
                "inputResponses": input_responses,
            });
            if let (Some(obj), Some(args)) = (extra.as_object_mut(), arguments.clone()) {
                obj.insert("arguments".to_string(), args);
            }
            if let (Some(obj), Some(state)) = (extra.as_object_mut(), request_state) {
                obj.insert("requestState".to_string(), Value::String(state));
            }
            return self
                .send_2026_07_28(
                    "prompts/get",
                    extra,
                    crate::protocol::v2026_07_28::parse_get_prompt,
                )
                .await;
        }

        let _ = (name, arguments, input_responses, request_state);
        Err(crate::error::ProtocolError::MethodNotFound(
            "MRTR input responses require a 2026-07-28 connection".to_string(),
        )
        .into())
    }

    /// `tools/call` that accepts EITHER a synchronous completion or a
    /// `CreateTaskResult` (`resultType: "task"`, SEP-2663). Requires a
    /// 2026-07-28 connection and `declared_capabilities.ext_tasks = true`
    /// (servers MUST NOT return task handles to clients that did not declare
    /// the extension).
    #[cfg(feature = "ext-tasks")]
    pub async fn call_tool_or_task(
        &self,
        name: &str,
        arguments: Value,
    ) -> McpClientResult<ToolCallOutcome> {
        if self.negotiated_version().await != Some(crate::version::McpVersion::V2026_07_28) {
            return Err(crate::error::ProtocolError::MethodNotFound(
                "the Tasks extension requires a 2026-07-28 connection".to_string(),
            )
            .into());
        }
        let extra = json!({ "name": name, "arguments": arguments });
        // No explicit Mcp-Name: the transport's apply_request_metadata_headers
        // derives it from params.name for tools/call and Base64-sentinel-encodes
        // it. Passing a raw one here would emit a second, unencoded Mcp-Name
        // header (reqwest appends rather than replaces).
        self.send_2026_07_28_with_extra_headers("tools/call", extra, &[], |result| {
            if result.get("resultType").and_then(|v| v.as_str())
                == Some(turul_mcp_ext_tasks::RESULT_TYPE_TASK)
            {
                let task: turul_mcp_ext_tasks::CreateTaskResult =
                    serde_json::from_value(result.clone())?;
                return Ok(ToolCallOutcome::Task(task));
            }
            crate::protocol::v2026_07_28::parse_call_tool(result).map(ToolCallOutcome::Completed)
        })
        .await
    }

    /// The MRTR retry that may be answered with a task — SEP-2663
    /// §Composition with Multi Round-Trip Requests.
    ///
    /// [`Self::call_tool_with_input_responses`] cannot express this: it routes
    /// through the strict `resultType` check, which rejects `"task"` as
    /// unrecognized. That check is right for a client that never declared the
    /// extension ("a resultType of any value unrecognized by the client MUST
    /// be considered invalid") and wrong for one that did — a server is
    /// explicitly permitted to gather input over MRTR and then hand back a
    /// `CreateTaskResult` on the final round. Without this method a client
    /// could not consume that composition at all, which an e2e against a
    /// server implementing it is the only thing that surfaces.
    #[cfg(feature = "ext-tasks")]
    pub async fn call_tool_or_task_with_input_responses(
        &self,
        name: &str,
        arguments: Value,
        input_responses: Value,
        request_state: Option<String>,
    ) -> McpClientResult<ToolCallOutcome> {
        if self.negotiated_version().await != Some(crate::version::McpVersion::V2026_07_28) {
            return Err(crate::error::ProtocolError::MethodNotFound(
                "the Tasks extension requires a 2026-07-28 connection".to_string(),
            )
            .into());
        }
        let mut extra = json!({
            "name": name,
            "arguments": arguments,
            "inputResponses": input_responses,
        });
        if let (Some(obj), Some(state)) = (extra.as_object_mut(), request_state) {
            obj.insert("requestState".to_string(), Value::String(state));
        }
        self.send_2026_07_28_with_extra_headers("tools/call", extra, &[], |result| {
            if result.get("resultType").and_then(|v| v.as_str())
                == Some(turul_mcp_ext_tasks::RESULT_TYPE_TASK)
            {
                let task: turul_mcp_ext_tasks::CreateTaskResult =
                    serde_json::from_value(result.clone())?;
                return Ok(ToolCallOutcome::Task(task));
            }
            crate::protocol::v2026_07_28::parse_call_tool(result).map(ToolCallOutcome::Completed)
        })
        .await
    }

    /// `tasks/get` — poll one task's state (SEP-2663).
    #[cfg(feature = "ext-tasks")]
    pub async fn task_get(
        &self,
        task_id: &str,
    ) -> McpClientResult<turul_mcp_ext_tasks::DetailedTask> {
        self.send_2026_07_28("tasks/get", json!({ "taskId": task_id }), |result| {
            let r: turul_mcp_ext_tasks::GetTaskResult = serde_json::from_value(result.clone())?;
            Ok(r.task)
        })
        .await
    }

    /// `tasks/update` — deliver input responses to an `input_required` task
    /// (SEP-2663). `input_responses` is the `{key: InputResponse}` map whose
    /// keys answer the task's outstanding `inputRequests`.
    #[cfg(feature = "ext-tasks")]
    pub async fn task_update(&self, task_id: &str, input_responses: Value) -> McpClientResult<()> {
        self.send_2026_07_28(
            "tasks/update",
            json!({ "taskId": task_id, "inputResponses": input_responses }),
            |_| Ok(()),
        )
        .await
    }

    /// `tasks/cancel` — cooperative cancellation (SEP-2663); the ack does not
    /// guarantee a `cancelled` terminal status.
    #[cfg(feature = "ext-tasks")]
    pub async fn task_cancel(&self, task_id: &str) -> McpClientResult<()> {
        self.send_2026_07_28("tasks/cancel", json!({ "taskId": task_id }), |_| Ok(()))
            .await
    }

    /// Poll `tasks/get` until the task reaches a terminal status, honoring
    /// the server's `pollIntervalMs` hint (clamped to [50ms, 30s]; default
    /// 500ms when the server sends none).
    #[cfg(feature = "ext-tasks")]
    pub async fn task_wait(
        &self,
        task_id: &str,
    ) -> McpClientResult<turul_mcp_ext_tasks::DetailedTask> {
        loop {
            let task = self.task_get(task_id).await?;
            if task.status().is_terminal() {
                return Ok(task);
            }
            let interval_ms = task
                .fields()
                .poll_interval_ms
                .unwrap_or(500.0)
                .clamp(50.0, 30_000.0);
            tokio::time::sleep(std::time::Duration::from_millis(interval_ms as u64)).await;
        }
    }

    /// List available resources (returns cached result if available)
    ///
    /// Returns the FIRST page only — use
    /// [`list_resources_paginated`](Self::list_resources_paginated) to walk all pages.
    ///
    /// The cache is automatically invalidated when the server sends a
    /// `notifications/resources/list_changed` notification. Use
    /// [`refresh_resources`](Self::refresh_resources) to force a fresh fetch.
    pub async fn list_resources(&self) -> McpClientResult<Vec<Resource>> {
        {
            let cache = self.cached_resources.read().await;
            if let Some(ref resources) = *cache {
                debug!(count = resources.len(), "Returning cached resources");
                return Ok(resources.clone());
            }
        }

        let resources = self.fetch_resources().await?;
        {
            let mut cache = self.cached_resources.write().await;
            *cache = Some(resources.clone());
        }
        Ok(resources)
    }

    /// Force a fresh fetch of resources from the server, bypassing and updating the cache.
    pub async fn refresh_resources(&self) -> McpClientResult<Vec<Resource>> {
        let resources = self.fetch_resources().await?;
        {
            let mut cache = self.cached_resources.write().await;
            *cache = Some(resources.clone());
        }
        Ok(resources)
    }

    /// Fetch resources from the server (no cache interaction).
    async fn fetch_resources(&self) -> McpClientResult<Vec<Resource>> {
        debug!("Fetching resources from server");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return self
                .send_2026_07_28(
                    "resources/list",
                    json!({}),
                    crate::protocol::v2026_07_28::parse_list_resources,
                )
                .await;
        }

        let request = self.build_request("resources/list", json!({}));

        let response = self.send_request_internal(request).await?;
        let resources_response: ListResourcesResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = resources_response.resources.len(),
            "Retrieved resources"
        );
        Ok(resources_response.resources)
    }

    /// List available resources with pagination support
    pub async fn list_resources_paginated(
        &self,
        cursor: Option<Cursor>,
    ) -> McpClientResult<ListResourcesResult> {
        debug!("Listing resources with pagination");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let extra = match cursor {
                Some(ref c) => json!({ "cursor": c.as_str() }),
                None => json!({}),
            };
            return self
                .send_2026_07_28(
                    "resources/list",
                    extra,
                    crate::protocol::v2026_07_28::parse_list_resources_result,
                )
                .await;
        }

        let request_params = if let Some(cursor) = cursor {
            json!({ "cursor": cursor.as_str() })
        } else {
            json!({})
        };

        let request = self.build_request("resources/list", request_params);

        let response = self.send_request_internal(request).await?;
        let resources_response: ListResourcesResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = resources_response.resources.len(),
            has_cursor = resources_response.next_cursor.is_some(),
            "Retrieved resources with pagination"
        );
        Ok(resources_response)
    }

    /// Read a resource
    pub async fn read_resource(
        &self,
        uri: &str,
    ) -> McpClientResult<Vec<turul_mcp_protocol_2025_11_25::ResourceContent>> {
        debug!(uri = uri, "Reading resource");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let r = self
                .send_2026_07_28(
                    "resources/read",
                    json!({ "uri": uri }),
                    crate::protocol::v2026_07_28::parse_read_resource,
                )
                .await?;
            return Ok(r.contents);
        }

        let request = self.build_request("resources/read", json!({ "uri": uri }));

        let response = self.send_request_internal(request).await?;
        let read_response: ReadResourceResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            uri = uri,
            content_count = read_response.contents.len(),
            "Resource read completed"
        );
        Ok(read_response.contents)
    }

    /// List available resource templates
    pub async fn list_resource_templates(&self) -> McpClientResult<Vec<ResourceTemplate>> {
        debug!("Listing resource templates");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return self
                .send_2026_07_28(
                    "resources/templates/list",
                    json!({}),
                    crate::protocol::v2026_07_28::parse_list_resource_templates,
                )
                .await;
        }

        let request = self.build_request("resources/templates/list", json!({}));

        let response = self.send_request_internal(request).await?;
        let templates_response: ListResourceTemplatesResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = templates_response.resource_templates.len(),
            "Retrieved resource templates"
        );
        Ok(templates_response.resource_templates)
    }

    /// List available resource templates with pagination support
    pub async fn list_resource_templates_paginated(
        &self,
        cursor: Option<Cursor>,
    ) -> McpClientResult<ListResourceTemplatesResult> {
        debug!("Listing resource templates with pagination");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let extra = match cursor {
                Some(ref c) => json!({ "cursor": c.as_str() }),
                None => json!({}),
            };
            return self
                .send_2026_07_28(
                    "resources/templates/list",
                    extra,
                    crate::protocol::v2026_07_28::parse_list_resource_templates_result,
                )
                .await;
        }

        let request_params = if let Some(cursor) = cursor {
            json!({ "cursor": cursor.as_str() })
        } else {
            json!({})
        };

        let request = self.build_request("resources/templates/list", request_params);

        let response = self.send_request_internal(request).await?;
        let templates_response: ListResourceTemplatesResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = templates_response.resource_templates.len(),
            has_cursor = templates_response.next_cursor.is_some(),
            "Retrieved resource templates with pagination"
        );
        Ok(templates_response)
    }

    /// List available prompts (returns cached result if available)
    ///
    /// Returns the FIRST page only — use
    /// [`list_prompts_paginated`](Self::list_prompts_paginated) to walk all pages.
    ///
    /// The cache is automatically invalidated when the server sends a
    /// `notifications/prompts/list_changed` notification. Use
    /// [`refresh_prompts`](Self::refresh_prompts) to force a fresh fetch.
    pub async fn list_prompts(&self) -> McpClientResult<Vec<Prompt>> {
        {
            let cache = self.cached_prompts.read().await;
            if let Some(ref prompts) = *cache {
                debug!(count = prompts.len(), "Returning cached prompts");
                return Ok(prompts.clone());
            }
        }

        let prompts = self.fetch_prompts().await?;
        {
            let mut cache = self.cached_prompts.write().await;
            *cache = Some(prompts.clone());
        }
        Ok(prompts)
    }

    /// Force a fresh fetch of prompts from the server, bypassing and updating the cache.
    pub async fn refresh_prompts(&self) -> McpClientResult<Vec<Prompt>> {
        let prompts = self.fetch_prompts().await?;
        {
            let mut cache = self.cached_prompts.write().await;
            *cache = Some(prompts.clone());
        }
        Ok(prompts)
    }

    /// Fetch prompts from the server (no cache interaction).
    async fn fetch_prompts(&self) -> McpClientResult<Vec<Prompt>> {
        debug!("Fetching prompts from server");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return self
                .send_2026_07_28(
                    "prompts/list",
                    json!({}),
                    crate::protocol::v2026_07_28::parse_list_prompts,
                )
                .await;
        }

        let request = self.build_request("prompts/list", json!({}));

        let response = self.send_request_internal(request).await?;
        let prompts_response: ListPromptsResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(count = prompts_response.prompts.len(), "Retrieved prompts");
        Ok(prompts_response.prompts)
    }

    /// List available prompts with pagination support
    pub async fn list_prompts_paginated(
        &self,
        cursor: Option<Cursor>,
    ) -> McpClientResult<ListPromptsResult> {
        debug!("Listing prompts with pagination");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let extra = match cursor {
                Some(ref c) => json!({ "cursor": c.as_str() }),
                None => json!({}),
            };
            return self
                .send_2026_07_28(
                    "prompts/list",
                    extra,
                    crate::protocol::v2026_07_28::parse_list_prompts_result,
                )
                .await;
        }

        let request_params = if let Some(cursor) = cursor {
            json!({ "cursor": cursor.as_str() })
        } else {
            json!({})
        };

        let request = self.build_request("prompts/list", request_params);

        let response = self.send_request_internal(request).await?;
        let prompts_response: ListPromptsResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = prompts_response.prompts.len(),
            has_cursor = prompts_response.next_cursor.is_some(),
            "Retrieved prompts with pagination"
        );
        Ok(prompts_response)
    }

    /// Get a prompt
    pub async fn get_prompt(
        &self,
        name: &str,
        arguments: Option<Value>,
    ) -> McpClientResult<GetPromptResult> {
        debug!(prompt = name, "Getting prompt");

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            let mut extra = json!({ "name": name });
            if let Some(ref args) = arguments {
                extra["arguments"] = args.clone();
            }
            return self
                .send_2026_07_28(
                    "prompts/get",
                    extra,
                    crate::protocol::v2026_07_28::parse_get_prompt,
                )
                .await;
        }

        let mut params = json!({
            "name": name
        });

        if let Some(args) = arguments {
            params["arguments"] = args;
        }

        let request = self.build_request("prompts/get", params);

        let response = self.send_request_internal(request).await?;
        let prompt_response: GetPromptResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            prompt = name,
            message_count = prompt_response.messages.len(),
            "Prompt retrieved"
        );
        Ok(prompt_response)
    }

    /// Ask the server to complete an argument value (`completion/complete`).
    ///
    /// `reference` names the prompt (`ref/prompt`) or resource template
    /// (`ref/resource`) the argument belongs to; `context` carries the values
    /// already resolved for the other arguments, which the server MAY use to
    /// narrow its suggestions.
    pub async fn complete(
        &self,
        reference: CompletionReference,
        argument: CompleteArgument,
        context: Option<CompletionContext>,
    ) -> McpClientResult<CompleteResult> {
        debug!(argument = %argument.name, "Requesting completions");

        let mut params = json!({
            "ref": serde_json::to_value(&reference)?,
            "argument": serde_json::to_value(&argument)?,
        });
        if let Some(ref context) = context {
            params["context"] = serde_json::to_value(context)?;
        }

        #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
        if self.negotiated_version().await == Some(crate::version::McpVersion::V2026_07_28) {
            return self
                .send_2026_07_28(
                    "completion/complete",
                    params,
                    crate::protocol::v2026_07_28::parse_complete,
                )
                .await;
        }

        let request = self.build_request("completion/complete", params);
        let response = self.send_request_internal(request).await?;
        let result: CompleteResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = result.completion.values.len(),
            "Completions received"
        );
        Ok(result)
    }

    /// Tell the server to stop working on an in-flight request
    /// (`notifications/cancelled`).
    ///
    /// `request_id` MUST be the JSON-RPC id of a request this client previously
    /// issued and that SHOULD still be in flight — e.g.
    /// [`SubscriptionStream::request_id`](SubscriptionStream::request_id) for a
    /// live `subscriptions/listen` stream. The notification is fire-and-forget:
    /// the server MAY still answer the original request if the cancellation
    /// arrives too late.
    ///
    /// Dropping a request future also closes its stream, which the server
    /// treats as cancellation; this is the explicit form, and the only one that
    /// can carry a `reason`.
    pub async fn cancel_request(
        &self,
        request_id: &str,
        reason: Option<&str>,
    ) -> McpClientResult<()> {
        debug!(request_id, "Cancelling request");

        let mut params = json!({ "requestId": request_id });
        if let Some(reason) = reason {
            params["reason"] = json!(reason);
        }

        self.send_notification_internal(Self::build_notification("notifications/cancelled", params))
            .await
    }

    /// Send a ping to test connectivity
    pub async fn ping(&self) -> McpClientResult<()> {
        debug!("Sending ping");

        self.reject_if_2026_07_28("ping").await?;

        let request = self.build_request("ping", json!({}));

        self.send_request_internal(request).await?;
        debug!("Ping successful");
        Ok(())
    }

    // === Task Operations ===

    /// Get a task by ID
    pub async fn get_task(&self, task_id: &str) -> McpClientResult<Task> {
        debug!(task_id = task_id, "Getting task");

        self.reject_if_2026_07_28("tasks/get").await?;

        let request = self.build_request("tasks/get", json!({ "taskId": task_id }));

        let response = self.send_request_internal(request).await?;
        let task_response: GetTaskResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(task_id = task_id, status = ?task_response.task.status, "Task retrieved");
        Ok(task_response.task)
    }

    /// List tasks
    pub async fn list_tasks(&self) -> McpClientResult<Vec<Task>> {
        debug!("Listing tasks");

        self.reject_if_2026_07_28("tasks/list").await?;

        let request = self.build_request("tasks/list", json!({}));

        let response = self.send_request_internal(request).await?;
        let tasks_response: ListTasksResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(count = tasks_response.tasks.len(), "Retrieved tasks");
        Ok(tasks_response.tasks)
    }

    /// List tasks with pagination support
    pub async fn list_tasks_paginated(
        &self,
        cursor: Option<Cursor>,
    ) -> McpClientResult<ListTasksResult> {
        debug!("Listing tasks with pagination");

        self.reject_if_2026_07_28("tasks/list").await?;

        let request_params = if let Some(cursor) = cursor {
            json!({ "cursor": cursor.as_str() })
        } else {
            json!({})
        };

        let request = self.build_request("tasks/list", request_params);

        let response = self.send_request_internal(request).await?;
        let tasks_response: ListTasksResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(
            count = tasks_response.tasks.len(),
            has_cursor = tasks_response.next_cursor.is_some(),
            "Retrieved tasks with pagination"
        );
        Ok(tasks_response)
    }

    /// Cancel a task
    pub async fn cancel_task(&self, task_id: &str) -> McpClientResult<Task> {
        debug!(task_id = task_id, "Cancelling task");

        self.reject_if_2026_07_28("tasks/cancel").await?;

        let request = self.build_request("tasks/cancel", json!({ "taskId": task_id }));

        let response = self.send_request_internal(request).await?;
        let cancel_response: CancelTaskResult =
            serde_json::from_value(response.get("result").cloned().unwrap_or(Value::Null))?;

        debug!(task_id = task_id, status = ?cancel_response.task.status, "Task cancelled");
        Ok(cancel_response.task)
    }

    /// Get a task's result (blocks until the task reaches terminal status)
    ///
    /// Per MCP spec, if the task is still in progress the server will hold the
    /// response until it completes. Use a longer timeout for this operation.
    pub async fn get_task_result(&self, task_id: &str) -> McpClientResult<Value> {
        debug!(task_id = task_id, "Getting task result");

        self.reject_if_2026_07_28("tasks/result").await?;

        let request = self.build_request("tasks/result", json!({ "taskId": task_id }));

        // Use long_operation timeout since tasks/result blocks until terminal
        let response = timeout(
            self.config.timeouts.long_operation,
            self.send_request_internal(request),
        )
        .await
        .map_err(|_| McpClientError::Timeout)??;

        let result = response.get("result").cloned().unwrap_or(Value::Null);
        debug!(task_id = task_id, "Task result retrieved");
        Ok(result)
    }

    /// Call a tool with task augmentation
    ///
    /// If the server supports tasks for this tool, it returns a `Task` (the tool
    /// executes asynchronously). Otherwise, it returns the normal `CallToolResult`.
    pub async fn call_tool_with_task(
        &self,
        name: &str,
        arguments: Value,
        ttl_ms: Option<i64>,
    ) -> McpClientResult<ToolCallResponse> {
        debug!(tool = name, "Calling tool with task augmentation");

        self.reject_if_2026_07_28("tasks (task-augmented tools/call)")
            .await?;

        let mut params = json!({
            "name": name,
            "arguments": arguments,
            "task": {}
        });

        if let Some(ttl) = ttl_ms {
            params["task"]["ttl"] = json!(ttl);
        }

        let request = self.build_request("tools/call", params);

        let response = self.send_request_internal(request).await?;
        let result = response.get("result").cloned().unwrap_or(Value::Null);

        // Distinguish response type: CreateTaskResult has a "task" field,
        // CallToolResult has a "content" field
        if result.get("task").is_some() {
            let task_result: CreateTaskResult = serde_json::from_value(result)?;
            debug!(
                tool = name,
                task_id = task_result.task.task_id,
                "Tool call created task"
            );
            Ok(ToolCallResponse::TaskCreated(task_result.task))
        } else {
            let call_result: CallToolResult = serde_json::from_value(result)?;
            debug!(
                tool = name,
                is_error = call_result.is_error,
                "Tool call completed synchronously"
            );
            Ok(ToolCallResponse::Immediate(call_result))
        }
    }

    /// Get stream handler for event callbacks
    pub async fn stream_handler(&self) -> tokio::sync::MutexGuard<'_, StreamHandler> {
        self.stream_handler.lock().await
    }

    /// Invalidate all cached lists (tools, resources, prompts).
    ///
    /// The next call to `list_tools()`, `list_resources()`, or `list_prompts()`
    /// will fetch fresh data from the server.
    pub async fn invalidate_caches(&self) {
        *self.cached_tools.write().await = None;
        *self.cached_resources.write().await = None;
        *self.cached_prompts.write().await = None;
        debug!("All list caches invalidated");
    }

    /// Get session information
    pub async fn session_info(&self) -> crate::session::SessionInfo {
        self.session.session_info().await
    }

    /// Get transport statistics
    pub async fn transport_stats(&self) -> crate::transport::TransportStatistics {
        self.transport.statistics()
    }
}

/// Connection status information
#[derive(Debug, Clone)]
pub struct ConnectionStatus {
    pub transport_connected: bool,
    pub session_state: SessionState,
    pub transport_type: crate::transport::TransportType,
    pub endpoint: String,
    pub session_id: Option<String>,
    pub protocol_version: Option<String>,
}

impl ConnectionStatus {
    /// Check if fully connected and ready
    pub fn is_ready(&self) -> bool {
        self.transport_connected && matches!(self.session_state, SessionState::Active)
    }

    /// Get status summary
    pub fn summary(&self) -> String {
        let session_display = match &self.session_id {
            Some(id) => &id[..id.len().min(8)],
            None => "None",
        };
        format!(
            "{} transport to {} - Session {} ({})",
            self.transport_type, self.endpoint, session_display, self.session_state
        )
    }
}

/// Response from a task-augmented tool call
///
/// When calling a tool with task augmentation, the server may either:
/// - Create a task and return it immediately (tool runs async)
/// - Execute the tool synchronously (if tasks not supported for this tool)
#[derive(Debug)]
pub enum ToolCallResponse {
    /// The tool executed synchronously and returned its result directly
    Immediate(CallToolResult),
    /// The server created a task — poll with `get_task()` or await with `get_task_result()`
    TaskCreated(Task),
}

impl ToolCallResponse {
    /// Returns `true` if the server created a task for this call
    pub fn is_task(&self) -> bool {
        matches!(self, ToolCallResponse::TaskCreated(_))
    }

    /// Returns the task if one was created
    pub fn task(&self) -> Option<&Task> {
        match self {
            ToolCallResponse::TaskCreated(task) => Some(task),
            _ => None,
        }
    }

    /// Returns the immediate result if the tool executed synchronously
    pub fn immediate_result(&self) -> Option<&CallToolResult> {
        match self {
            ToolCallResponse::Immediate(result) => Some(result),
            _ => None,
        }
    }
}

/// Builder for creating MCP clients
pub struct McpClientBuilder {
    transport: Option<BoxedTransport>,
    url: Option<String>,
    config: Option<ClientConfig>,
    notification_callback: Option<NotificationCallback>,
}

impl McpClientBuilder {
    /// Create a new client builder
    pub fn new() -> Self {
        Self {
            transport: None,
            url: None,
            config: None,
            notification_callback: None,
        }
    }

    /// Set transport directly (ConnectionConfig is NOT applied — caller owns the transport)
    pub fn with_transport(mut self, transport: BoxedTransport) -> Self {
        self.transport = Some(transport);
        self.url = None; // explicit transport overrides URL
        self
    }

    /// Set transport endpoint URL (transport constructed lazily in `build()` with config applied)
    ///
    /// Can be called before or after `with_config()` — config is applied at build time.
    pub fn with_url(mut self, url: &str) -> McpClientResult<Self> {
        // Validate URL eagerly so errors surface at call site
        let parsed = url::Url::parse(url).map_err(|e| {
            crate::error::TransportError::ConnectionFailed(format!("Invalid URL: {}", e))
        })?;
        if !matches!(parsed.scheme(), "http" | "https") {
            return Err(crate::error::TransportError::ConnectionFailed(format!(
                "Invalid scheme: {}",
                parsed.scheme()
            ))
            .into());
        }
        self.url = Some(url.to_string());
        self.transport = None; // URL overrides explicit transport
        Ok(self)
    }

    /// Set configuration (applied to transport at build time if using `with_url`)
    pub fn with_config(mut self, config: ClientConfig) -> Self {
        self.config = Some(config);
        self
    }

    /// Register a callback for server notifications.
    ///
    /// The callback fires for every server notification, including
    /// `notifications/tools/list_changed`, `notifications/resources/list_changed`,
    /// and `notifications/prompts/list_changed`.
    ///
    /// Note: The built-in cache invalidation for `list_changed` notifications
    /// always runs regardless of whether a user callback is registered.
    pub fn on_notification<F>(mut self, callback: F) -> Self
    where
        F: Fn(&str, Option<&Value>) + Send + Sync + 'static,
    {
        self.notification_callback = Some(Arc::new(callback));
        self
    }

    /// Build the client
    ///
    /// If `with_url()` was used, the transport is constructed here with `ConnectionConfig` applied.
    /// If `with_transport()` was used, the provided transport is used as-is.
    pub fn build(self) -> McpClient {
        let config = self.config.unwrap_or_default();
        let transport = if let Some(transport) = self.transport {
            transport
        } else if let Some(ref url) = self.url {
            // Detect transport type from URL, then construct with config applied
            let transport_type = crate::transport::detect_transport_type(url)
                .expect("URL was validated in with_url() but detection failed");
            match transport_type {
                crate::transport::TransportType::Http => Box::new(
                    crate::transport::http::HttpTransport::with_config(url, &config.connection)
                        .expect(
                            "URL was validated in with_url() but transport construction failed",
                        ),
                )
                    as crate::transport::BoxedTransport,
                crate::transport::TransportType::Sse => {
                    // SSE is a legacy transport — ConnectionConfig not wired (no with_config)
                    Box::new(
                        #[allow(deprecated)] // ≤2024-11-05 fallback (SEP-2596)
                        crate::transport::sse::SseTransport::new(url)
                            .expect("URL was validated in with_url() but SSE construction failed"),
                    )
                }
            }
        } else {
            panic!("Transport must be set via with_transport() or with_url() before building");
        };

        McpClient::new_with_callback(transport, config, self.notification_callback)
    }
}

impl Default for McpClientBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Convert a `serde_json::Value` to `Option<RequestParams>`, preserving the
/// invariant that `Value::Object(empty)` becomes `Some(empty)` (renders as
/// `"params":{}`), not `None` (which would omit the field via
/// `skip_serializing_if`). All MCP client call sites pass either an Object
/// (for named params) or `Value::Null` (no params); Array params are accepted
/// for completeness but no current call site uses them.
fn value_to_request_params(params: Value) -> Option<RequestParams> {
    match params {
        Value::Null => None,
        Value::Object(map) => Some(RequestParams::Object(map.into_iter().collect())),
        Value::Array(arr) => Some(RequestParams::Array(arr)),
        other => unreachable!(
            "MCP client requests use object or null params; got scalar: {:?}",
            other
        ),
    }
}

/// Outcome of [`McpClient::call_tool_or_task`] (SEP-2663): the server is the
/// sole decider of task materialization.
#[cfg(feature = "ext-tasks")]
#[derive(Debug)]
pub enum ToolCallOutcome {
    /// The call completed synchronously with an ordinary tool result.
    Completed(CallToolResult),
    /// The server elected a task — poll with
    /// [`McpClient::task_wait`]/[`McpClient::task_get`].
    Task(turul_mcp_ext_tasks::CreateTaskResult),
}

/// A live `subscriptions/listen` stream: the acknowledged filter subset, the
/// server-assigned subscription id, and the ordered notification feed.
/// Dropping this closes the stream (= cancels the subscription).
pub struct SubscriptionStream {
    /// The filter subset the server agreed to honor (the acknowledgement's
    /// `params.notifications`, verbatim).
    pub honored: Value,
    /// `io.modelcontextprotocol/subscriptionId` from the acknowledgement.
    pub subscription_id: Option<String>,
    /// JSON-RPC id of the `subscriptions/listen` request that opened this
    /// stream — the id [`McpClient::cancel_request`] takes to close it
    /// explicitly.
    request_id: Option<String>,
    receiver: tokio::sync::mpsc::UnboundedReceiver<Value>,
}

impl SubscriptionStream {
    /// Next notification on the stream (`None` when the server closes it).
    pub async fn next(&mut self) -> Option<Value> {
        self.receiver.recv().await
    }

    /// JSON-RPC id of the `subscriptions/listen` request that opened this
    /// stream. Pass it to [`McpClient::cancel_request`] to close the stream
    /// with a reason instead of dropping it.
    pub fn request_id(&self) -> Option<&str> {
        self.request_id.as_deref()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::http::HttpTransport;
    use crate::transport::{
        ConnectionInfo, EventReceiver, ServerEvent, TransportCapabilities, TransportResponse,
        TransportStatistics, TransportType,
    };
    use async_trait::async_trait;
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
    use tokio::sync::mpsc;

    #[tokio::test]
    async fn test_client_builder() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();
        let client = McpClientBuilder::new()
            .with_transport(Box::new(transport))
            .build();

        // Basic smoke test
        assert!(!client.is_ready().await);
    }

    #[test]
    fn test_connection_status() {
        let status = ConnectionStatus {
            transport_connected: true,
            session_state: SessionState::Active,
            transport_type: crate::transport::TransportType::Http,
            endpoint: "http://localhost:8080/mcp".to_string(),
            session_id: Some("session123".to_string()),
            protocol_version: Some("2025-11-25".to_string()),
        };

        assert!(status.is_ready());
        assert!(status.summary().contains("HTTP transport"));
    }

    /// Mock transport that records send_notification() calls and provides
    /// a controllable event channel for injecting ServerEvents.
    struct MockTransport {
        #[cfg_attr(
            not(any(feature = "client-bilingual", feature = "client-2025-11-25-only")),
            allow(dead_code)
        )]
        event_tx: mpsc::UnboundedSender<ServerEvent>,
        event_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<ServerEvent>>>,
        notifications: Arc<tokio::sync::Mutex<Vec<Value>>>,
        connected: AtomicBool,
        /// Count of `send_delete` calls — observable from the test side via a
        /// pre-cloned Arc, since the transport is moved into `Box` before connect.
        delete_count: Arc<std::sync::atomic::AtomicU32>,
    }

    impl MockTransport {
        fn new() -> (Self, Arc<tokio::sync::Mutex<Vec<Value>>>) {
            let (event_tx, event_rx) = mpsc::unbounded_channel();
            let notifications = Arc::new(tokio::sync::Mutex::new(Vec::new()));
            let mock = Self {
                event_tx,
                event_rx: parking_lot::Mutex::new(Some(event_rx)),
                notifications: Arc::clone(&notifications),
                connected: AtomicBool::new(false),
                delete_count: Arc::new(std::sync::atomic::AtomicU32::new(0)),
            };
            (mock, notifications)
        }

        /// Get the event sender for injecting server events from the test
        #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
        fn event_sender(&self) -> mpsc::UnboundedSender<ServerEvent> {
            self.event_tx.clone()
        }

        /// Pre-clone the DELETE-count handle so a test can observe it after the
        /// transport has been boxed and moved into the client.
        #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
        fn delete_count_handle(&self) -> Arc<std::sync::atomic::AtomicU32> {
            Arc::clone(&self.delete_count)
        }
    }

    #[async_trait]
    impl crate::transport::Transport for MockTransport {
        fn transport_type(&self) -> TransportType {
            TransportType::Http
        }

        fn capabilities(&self) -> TransportCapabilities {
            TransportCapabilities {
                streaming: true,
                bidirectional: true,
                server_events: true,
                max_message_size: None,
                persistent: true,
            }
        }

        async fn connect(&self) -> McpClientResult<()> {
            self.connected.store(true, AtomicOrdering::SeqCst);
            Ok(())
        }

        async fn disconnect(&self) -> McpClientResult<()> {
            self.connected.store(false, AtomicOrdering::SeqCst);
            Ok(())
        }

        fn is_connected(&self) -> bool {
            self.connected.load(AtomicOrdering::SeqCst)
        }

        async fn send_request(&self, _request: Value) -> McpClientResult<Value> {
            // Not used in this test path
            Ok(json!({"jsonrpc": "2.0", "result": {}}))
        }

        async fn send_request_with_headers(
            &self,
            request: Value,
        ) -> McpClientResult<TransportResponse> {
            // A real 2025-11-25 server answers `server/discover` with JSON-RPC
            // -32601, which drives the bilingual client to fall back to `initialize`.
            if request.get("method").and_then(|m| m.as_str()) == Some("server/discover") {
                return Ok(TransportResponse::new(
                    json!({"jsonrpc": "2.0", "id": "req_0", "error": {"code": -32601, "message": "Method not found"}}),
                    HashMap::new(),
                ));
            }
            // Return a valid initialize response with session ID
            let mut headers = HashMap::new();
            headers.insert("mcp-session-id".to_string(), "mock-session-123".to_string());

            Ok(TransportResponse::new(
                json!({
                    "jsonrpc": "2.0",
                    "id": "req_0",
                    "result": {
                        "protocolVersion": "2025-11-25",
                        "capabilities": {
                            // listChanged is false: this static MockTransport does NOT
                            // emit notifications/tools/list_changed during initialize, so
                            // advertising true would violate MCP capability truthfulness.
                            // Tests that exercise list_changed inject the notification
                            // via `event_sender()` (an out-of-band channel), independent
                            // of the advertised capability.
                            "tools": { "listChanged": false }
                        },
                        "serverInfo": {
                            "name": "mock-server",
                            "version": "1.0.0"
                        }
                    }
                }),
                headers,
            ))
        }

        async fn send_notification(&self, notification: Value) -> McpClientResult<()> {
            self.notifications.lock().await.push(notification);
            Ok(())
        }

        async fn send_delete(&self, _session_id: &str) -> McpClientResult<()> {
            self.delete_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }

        fn set_session_id(&self, _session_id: String) {}

        fn clear_session_id(&self) {}

        async fn start_event_listener(&self) -> McpClientResult<EventReceiver> {
            self.event_rx
                .lock()
                .take()
                .ok_or_else(|| McpClientError::generic("Event listener already started"))
        }

        fn connection_info(&self) -> ConnectionInfo {
            ConnectionInfo {
                transport_type: TransportType::Http,
                endpoint: "mock://test".to_string(),
                connected: self.is_connected(),
                capabilities: self.capabilities(),
                metadata: Value::Null,
            }
        }

        fn statistics(&self) -> TransportStatistics {
            TransportStatistics::default()
        }
    }

    /// Verifies the full McpClient pipeline: server request → StreamHandler callback
    /// → response channel → consumer task → transport.send_notification().
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_client_response_consumer_pipeline() {
        let (mock, notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();

        // connect() wires up StreamHandler + consumer task + runs initialization
        client.connect().await.unwrap();

        // Register a request callback via the public API
        {
            let handler = client.stream_handler().await;
            handler.on_request(|request| {
                let method = request.get("method").and_then(|m| m.as_str()).unwrap_or("");
                match method {
                    "sampling/createMessage" => Ok(json!({
                        "role": "assistant",
                        "content": { "type": "text", "text": "mock response" },
                        "model": "test-model"
                    })),
                    _ => Err(format!("Unsupported: {}", method)),
                }
            });
        }

        // Inject a server-initiated request through the event channel
        event_sender
            .send(ServerEvent::Request(json!({
                "jsonrpc": "2.0",
                "id": "srv-req-42",
                "method": "sampling/createMessage",
                "params": {
                    "messages": [{"role": "user", "content": {"type": "text", "text": "Hi"}}],
                    "maxTokens": 100
                }
            })))
            .unwrap();

        // Wait for the consumer task to forward the response to transport
        let response = tokio::time::timeout(std::time::Duration::from_secs(2), async {
            loop {
                let notifs = notifications.lock().await;
                // Skip the first notification (notifications/initialized from connect())
                let responses: Vec<&Value> =
                    notifs.iter().filter(|n| n.get("id").is_some()).collect();
                if !responses.is_empty() {
                    return responses[0].clone();
                }
                drop(notifs);
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("timed out waiting for response to reach transport");

        // Verify JSON-RPC 2.0 response structure
        assert_eq!(response["jsonrpc"], "2.0");
        assert_eq!(response["id"], "srv-req-42");
        assert!(
            response.get("error").is_none(),
            "should not have error field"
        );
        assert_eq!(response["result"]["role"], "assistant");
        assert_eq!(response["result"]["model"], "test-model");

        client.disconnect().await.unwrap();
    }

    /// Same pipeline but with an error callback — verifies error responses reach transport.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_client_response_consumer_pipeline_error() {
        let (mock, notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();

        client.connect().await.unwrap();

        {
            let handler = client.stream_handler().await;
            handler.on_request(|_req| Err("not supported".to_string()));
        }

        event_sender
            .send(ServerEvent::Request(json!({
                "jsonrpc": "2.0",
                "id": 99,
                "method": "elicitation/create",
                "params": {}
            })))
            .unwrap();

        let response = tokio::time::timeout(std::time::Duration::from_secs(2), async {
            loop {
                let notifs = notifications.lock().await;
                let responses: Vec<&Value> =
                    notifs.iter().filter(|n| n.get("id").is_some()).collect();
                if !responses.is_empty() {
                    return responses[0].clone();
                }
                drop(notifs);
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("timed out waiting for error response");

        assert_eq!(response["jsonrpc"], "2.0");
        assert_eq!(response["id"], 99);
        assert_eq!(response["error"]["code"], -32603);
        assert!(
            response["error"]["message"]
                .as_str()
                .unwrap()
                .contains("not supported")
        );

        client.disconnect().await.unwrap();
    }

    /// `disconnect()` must clear the SessionManager's session_id so that the
    /// subsequent `Drop` cleanup sees no ID and skips its DELETE. Without this
    /// idempotency the client fires two DELETEs (one from disconnect, one from
    /// Drop) — the second lands on a server with the session already gone, and
    /// in OAuth deployments may arrive after the bearer token has expired.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_disconnect_clears_session_so_drop_is_noop() {
        let (mock, _notifications) = MockTransport::new();
        let delete_count = mock.delete_count_handle();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();
        client.connect().await.unwrap();

        client.disconnect().await.unwrap();
        assert_eq!(
            delete_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "explicit disconnect must send exactly one DELETE"
        );

        // Drop the client — the cleanup task is spawned; give the runtime a
        // moment to schedule it. If the idempotency contract holds the
        // spawned task hits the `None` branch and returns without DELETE.
        drop(client);
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        assert_eq!(
            delete_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "Drop after disconnect must NOT send a second DELETE"
        );
    }

    /// Regression guard for the existing Drop-without-disconnect path: a caller
    /// that just lets the client fall out of scope (no explicit `disconnect()`)
    /// must still get the DELETE sent by Drop. Locking this in protects against
    /// future "always clear session_id eagerly" refactors that would silently
    /// break server-side session cleanup for the implicit-drop path.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_drop_without_disconnect_still_fires_delete() {
        let (mock, _notifications) = MockTransport::new();
        let delete_count = mock.delete_count_handle();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();
        client.connect().await.unwrap();

        drop(client);
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        assert_eq!(
            delete_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "Drop without prior disconnect must send DELETE"
        );
    }

    #[tokio::test]
    async fn test_builder_with_sse_url_yields_sse_transport() {
        let client = McpClientBuilder::new()
            .with_url("http://localhost:9999/sse")
            .unwrap()
            .build();
        let status = client.connection_status().await;
        assert_eq!(
            status.transport_type,
            crate::transport::TransportType::Sse,
            "URL with /sse path must yield SSE transport"
        );
    }

    #[tokio::test]
    async fn test_builder_with_mcp_url_yields_http_transport() {
        let client = McpClientBuilder::new()
            .with_url("http://localhost:9999/mcp")
            .unwrap()
            .build();
        let status = client.connection_status().await;
        assert_eq!(
            status.transport_type,
            crate::transport::TransportType::Http,
            "Non-SSE URL must yield HTTP transport"
        );
    }

    // ── StatefulMockTransport ──────────────────────────────────────────

    use std::collections::VecDeque;
    use std::sync::atomic::AtomicU32;

    /// A mock transport that supports multi-step response sequences and call tracking.
    /// Unlike `MockTransport` (which returns fixed responses), this transport pops
    /// responses from queues, enabling tests that exercise full session lifecycles
    /// (initialize → set session → requests → re-initialize on 404 → …).
    #[allow(dead_code)]
    struct StatefulMockTransport {
        /// Sequence of responses for send_request_with_headers (initialize)
        init_responses: Arc<std::sync::Mutex<VecDeque<McpClientResult<TransportResponse>>>>,
        /// Sequence of responses for send_request (normal requests)
        request_responses: Arc<std::sync::Mutex<VecDeque<McpClientResult<Value>>>>,
        /// Tracks set_session_id calls in order
        set_session_ids: Arc<std::sync::Mutex<Vec<String>>>,
        /// Tracks clear_session_id call count
        clear_count: Arc<AtomicU32>,
        /// Event channel sender
        event_tx: Option<mpsc::UnboundedSender<ServerEvent>>,
        /// Event channel receiver (taken once by start_event_listener)
        event_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<ServerEvent>>>,
        /// Capabilities
        caps: TransportCapabilities,
        /// Connected flag
        connected: AtomicBool,
    }

    impl StatefulMockTransport {
        fn new() -> Self {
            let (event_tx, event_rx) = mpsc::unbounded_channel();
            Self {
                init_responses: Arc::new(std::sync::Mutex::new(VecDeque::new())),
                request_responses: Arc::new(std::sync::Mutex::new(VecDeque::new())),
                set_session_ids: Arc::new(std::sync::Mutex::new(Vec::new())),
                clear_count: Arc::new(AtomicU32::new(0)),
                event_tx: Some(event_tx),
                event_rx: parking_lot::Mutex::new(Some(event_rx)),
                caps: TransportCapabilities {
                    streaming: true,
                    bidirectional: true,
                    server_events: true,
                    max_message_size: None,
                    persistent: true,
                },
                connected: AtomicBool::new(false),
            }
        }

        #[allow(dead_code)]
        fn push_init_response(&mut self, resp: McpClientResult<TransportResponse>) {
            self.init_responses.lock().unwrap().push_back(resp);
        }

        fn push_request_response(&mut self, resp: McpClientResult<Value>) {
            self.request_responses.lock().unwrap().push_back(resp);
        }

        /// Create a valid initialize TransportResponse with optional session ID header.
        #[allow(dead_code)]
        fn make_init_response(
            session_id: Option<&str>,
            protocol_version: &str,
        ) -> TransportResponse {
            let mut headers = HashMap::new();
            if let Some(sid) = session_id {
                headers.insert("mcp-session-id".to_string(), sid.to_string());
            }
            TransportResponse::new(
                json!({
                    "jsonrpc": "2.0",
                    "id": "req_0",
                    "result": {
                        "protocolVersion": protocol_version,
                        "capabilities": {
                            // listChanged is false: this StatefulMockTransport does NOT
                            // emit notifications/tools/list_changed during initialize, so
                            // advertising true would violate MCP capability truthfulness.
                            // Tests that exercise list_changed inject the notification
                            // via `event_sender()` (an out-of-band channel), independent
                            // of the advertised capability.
                            "tools": { "listChanged": false }
                        },
                        "serverInfo": {
                            "name": "stateful-mock",
                            "version": "1.0.0"
                        }
                    }
                }),
                headers,
            )
        }
    }

    #[async_trait]
    impl crate::transport::Transport for StatefulMockTransport {
        fn transport_type(&self) -> TransportType {
            TransportType::Http
        }

        fn capabilities(&self) -> TransportCapabilities {
            self.caps.clone()
        }

        async fn connect(&self) -> McpClientResult<()> {
            self.connected.store(true, AtomicOrdering::SeqCst);
            Ok(())
        }

        async fn disconnect(&self) -> McpClientResult<()> {
            self.connected.store(false, AtomicOrdering::SeqCst);
            Ok(())
        }

        fn is_connected(&self) -> bool {
            self.connected.load(AtomicOrdering::SeqCst)
        }

        async fn send_request(&self, _request: Value) -> McpClientResult<Value> {
            self.request_responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| {
                    Err(McpClientError::generic(
                        "StatefulMockTransport: no more request responses queued",
                    ))
                })
        }

        async fn send_request_with_headers(
            &self,
            request: Value,
        ) -> McpClientResult<TransportResponse> {
            // Answer the version probe like a 2025-11-25 server (-32601) without
            // consuming a queued init response, so fallback drives initialize.
            if request.get("method").and_then(|m| m.as_str()) == Some("server/discover") {
                return Ok(TransportResponse::new(
                    json!({"jsonrpc": "2.0", "id": "req_0", "error": {"code": -32601, "message": "Method not found"}}),
                    HashMap::new(),
                ));
            }
            self.init_responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| {
                    Err(McpClientError::generic(
                        "StatefulMockTransport: no more init responses queued",
                    ))
                })
        }

        async fn send_notification(&self, _notification: Value) -> McpClientResult<()> {
            Ok(())
        }

        async fn send_delete(&self, _session_id: &str) -> McpClientResult<()> {
            Ok(())
        }

        fn set_session_id(&self, session_id: String) {
            self.set_session_ids.lock().unwrap().push(session_id);
        }

        fn clear_session_id(&self) {
            self.clear_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        }

        async fn start_event_listener(&self) -> McpClientResult<EventReceiver> {
            self.event_rx
                .lock()
                .take()
                .ok_or_else(|| McpClientError::generic("Event listener already started"))
        }

        fn connection_info(&self) -> ConnectionInfo {
            ConnectionInfo {
                transport_type: TransportType::Http,
                endpoint: "stateful-mock://test".to_string(),
                connected: self.is_connected(),
                capabilities: self.caps.clone(),
                metadata: Value::Null,
            }
        }

        fn statistics(&self) -> TransportStatistics {
            TransportStatistics::default()
        }
    }

    #[tokio::test]
    async fn test_stateful_mock_transport_sequences() {
        use crate::transport::Transport;

        let mut transport = StatefulMockTransport::new();
        transport.push_request_response(Ok(json!({"result": "first"})));
        transport.push_request_response(Ok(json!({"result": "second"})));

        let r1 = transport.send_request(json!({})).await.unwrap();
        assert_eq!(r1["result"], "first");
        let r2 = transport.send_request(json!({})).await.unwrap();
        assert_eq!(r2["result"], "second");

        // Queue exhausted → error
        assert!(transport.send_request(json!({})).await.is_err());
    }

    // ── Session lifecycle tests ─────────────────────────────────────────

    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    use crate::config::RetryConfig;
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    use std::sync::atomic::Ordering;
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    use std::time::Duration;

    /// Helper: build a fast-retry config (1 ms delays) with the given max_attempts.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    fn fast_retry_config(max_attempts: u32) -> ClientConfig {
        ClientConfig {
            retry: RetryConfig {
                max_attempts,
                initial_delay: Duration::from_millis(1),
                max_delay: Duration::from_millis(1),
                ..Default::default()
            },
            ..Default::default()
        }
    }

    /// Test 2.1 — 404 recovery path resets session, clears stale transport session ID,
    /// re-initializes with a fresh session, and retries the original request.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_404_reinitialize_clears_stale_session_id() {
        let mut transport = StatefulMockTransport::new();

        // 1. Initial initialize → session "AAA"
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. First real request → 404 (session expired on server)
        transport.push_request_response(Err(McpClientError::Transport(
            crate::error::TransportError::HttpStatus {
                status: 404,
                message: "Not Found".to_string(),
            },
        )));

        // 3. Re-initialize → session "BBB"
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));

        // 4. Retry request → success
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": { "tools": [] }
        })));

        let clear_count = transport.clear_count.clone();
        let set_ids = transport.set_session_ids.clone();

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        // This should: fail with 404 → clear session → re-init → retry → succeed
        let result = client.list_tools().await;
        assert!(
            result.is_ok(),
            "Request should succeed after 404 re-initialization: {:?}",
            result.err()
        );

        // Verify clear_session_id was called exactly once during 404 recovery
        assert_eq!(
            clear_count.load(Ordering::SeqCst),
            1,
            "clear_session_id must be called exactly once during 404 recovery"
        );

        // Verify new session ID was set after re-initialization
        let ids = set_ids.lock().unwrap();
        assert!(
            ids.contains(&"session-BBB".to_string()),
            "New session ID 'session-BBB' should be set after re-initialization, got: {:?}",
            *ids
        );
    }

    /// Test 2.1a — 404 on last retry attempt still recovers (re-init doesn't count
    /// as a "retry" — the loop continues after successful re-init).
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_404_on_last_retry_attempt_still_recovers() {
        let mut transport = StatefulMockTransport::new();

        // Initial init
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));
        // 404 on first request
        transport.push_request_response(Err(McpClientError::Transport(
            crate::error::TransportError::HttpStatus {
                status: 404,
                message: "Not Found".to_string(),
            },
        )));
        // Re-init
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));
        // Retry succeeds
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": { "tools": [] }
        })));

        let client = McpClient::new(Box::new(transport), fast_retry_config(2));
        client.connect().await.unwrap();

        let result = client.list_tools().await;
        assert!(
            result.is_ok(),
            "404 recovery should work even with max_attempts=2: {:?}",
            result.err()
        );
    }

    /// Test 2.1b — When re-initialization after 404 fails, the original 404 error
    /// is surfaced (not the re-init error).
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_404_reinit_failure_surfaces_original_error() {
        let mut transport = StatefulMockTransport::new();

        // Initial init
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));
        // 404
        transport.push_request_response(Err(McpClientError::Transport(
            crate::error::TransportError::HttpStatus {
                status: 404,
                message: "Session gone".to_string(),
            },
        )));
        // Re-init FAILS
        transport.push_init_response(Err(McpClientError::Transport(
            crate::error::TransportError::ConnectionFailed("Connection refused".to_string()),
        )));

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        let result = client.list_tools().await;
        assert!(result.is_err());

        // Should surface the original 404 error, not the reinit ConnectionFailed error
        let err = result.unwrap_err();
        assert!(
            err.is_session_expired(),
            "Should surface original 404 error, got: {}",
            err
        );
    }

    /// Test 2.2 — Server may omit Mcp-Session-Id header (stateless mode).
    /// connect() must succeed and client must be ready.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_optional_session_id_no_hard_failure() {
        let mut transport = StatefulMockTransport::new();

        // Init response WITHOUT session ID
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            None, // No session ID
            "2025-11-25",
        )));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        let result = client.connect().await;
        assert!(
            result.is_ok(),
            "connect() must succeed without Mcp-Session-Id: {:?}",
            result.err()
        );
        assert!(
            client.is_ready().await,
            "Client must be ready after stateless init"
        );

        let status = client.connection_status().await;
        assert!(
            status.session_id.is_none(),
            "Session ID should be None for stateless server, got: {:?}",
            status.session_id
        );
    }

    /// Test 2.3 — Server returns an unsupported protocol version.
    /// connect() must fail with an error mentioning both versions.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_unsupported_protocol_version_rejected() {
        let mut transport = StatefulMockTransport::new();

        // Server returns unsupported version
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-123"),
            "2099-01-01",
        )));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        let result = client.connect().await;
        assert!(
            result.is_err(),
            "connect() must fail for unsupported protocol version"
        );

        let err = result.unwrap_err();
        let err_msg = err.to_string();
        assert!(
            err_msg.contains("2099-01-01"),
            "Error should mention server's version '2099-01-01', got: {}",
            err_msg
        );
        assert!(
            err_msg.contains("2025-11-25"),
            "Error should mention supported version '2025-11-25', got: {}",
            err_msg
        );
    }

    // ── Error propagation tests ────────────────────────────────────────

    /// Test 4.1 — JSON-RPC error response surfaces as `ServerError` with code, message, and data.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_jsonrpc_error_surfaces_as_server_error_with_code_message_data() {
        let mut transport = StatefulMockTransport::new();
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32602,
                "message": "Invalid params",
                "data": {"detail": "missing field 'name'"}
            }
        })));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let result = client.list_tools().await;
        assert!(result.is_err());

        let err = result.unwrap_err();
        assert_eq!(err.error_code(), Some(-32602));
        assert!(err.to_string().contains("Invalid params"));
        if let McpClientError::ServerError { data, .. } = &err {
            assert!(data.is_some());
            assert_eq!(data.as_ref().unwrap()["detail"], "missing field 'name'");
        } else {
            panic!("Expected ServerError, got: {:?}", err);
        }
    }

    /// Test 4.2 — JSON-RPC error without `data` field: `data` must be `None`.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_jsonrpc_error_without_data_field() {
        let mut transport = StatefulMockTransport::new();
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32600,
                "message": "Invalid Request"
            }
        })));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let result = client.list_tools().await;
        assert!(result.is_err());

        let err = result.unwrap_err();
        assert_eq!(err.error_code(), Some(-32600));
        if let McpClientError::ServerError { data, .. } = &err {
            assert!(data.is_none(), "data should be None when server omits it");
        } else {
            panic!("Expected ServerError, got: {:?}", err);
        }
    }

    /// Test 4.3 — `call_tool` with a malformed response returns an error rather than panicking.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_call_tool_malformed_response_returns_error() {
        let mut transport = StatefulMockTransport::new();
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": {"unexpected": "shape"}
        })));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let result = client.call_tool("test", json!({})).await;
        assert!(
            result.is_err(),
            "Malformed response should return error, not panic"
        );
    }

    /// Test 4.4 — `get_prompt` with a malformed response returns an error rather than panicking.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_get_prompt_malformed_response_returns_error() {
        let mut transport = StatefulMockTransport::new();
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": {"wrong": "format"}
        })));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let result = client.get_prompt("test", None).await;
        assert!(
            result.is_err(),
            "Malformed response should return error, not panic"
        );
    }

    // ── Cache and notification tests ─────────────────────────────────────

    /// Test: list_tools() caches results and returns cached data on second call.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_list_tools_caches_result() {
        let mut transport = StatefulMockTransport::new();

        // Init
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));

        // First list_tools call
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": {
                "tools": [
                    {"name": "tool_a", "description": "Tool A", "inputSchema": {"type": "object"}}
                ]
            }
        })));

        // No second response queued — if list_tools makes a second request, it will error

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let tools1 = client.list_tools().await.unwrap();
        assert_eq!(tools1.len(), 1);
        assert_eq!(tools1[0].name, "tool_a");

        // Second call should use cache (no network request)
        let tools2 = client.list_tools().await.unwrap();
        assert_eq!(tools2.len(), 1);
        assert_eq!(tools2[0].name, "tool_a");
    }

    /// Test: notifications/tools/list_changed invalidates the tool cache.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_tools_list_changed_notification_invalidates_cache() {
        let (mock, _notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();

        client.connect().await.unwrap();

        // Manually populate the tool cache
        {
            let mut cache = client.cached_tools.write().await;
            *cache = Some(vec![]);
        }

        // Verify cache is populated
        assert!(client.cached_tools.read().await.is_some());

        // Send notifications/tools/list_changed
        event_sender
            .send(ServerEvent::Notification(json!({
                "jsonrpc": "2.0",
                "method": "notifications/tools/list_changed"
            })))
            .unwrap();

        // Wait for the notification to be processed
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Cache should be invalidated
        assert!(
            client.cached_tools.read().await.is_none(),
            "Tool cache should be invalidated after notifications/tools/list_changed"
        );
    }

    /// Test: notifications/resources/list_changed invalidates the resource cache.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_resources_list_changed_notification_invalidates_cache() {
        let (mock, _notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();

        client.connect().await.unwrap();

        // Populate resource cache
        {
            let mut cache = client.cached_resources.write().await;
            *cache = Some(vec![]);
        }

        assert!(client.cached_resources.read().await.is_some());

        event_sender
            .send(ServerEvent::Notification(json!({
                "jsonrpc": "2.0",
                "method": "notifications/resources/list_changed"
            })))
            .unwrap();

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

        assert!(
            client.cached_resources.read().await.is_none(),
            "Resource cache should be invalidated after notifications/resources/list_changed"
        );
    }

    /// Test: notifications/prompts/list_changed invalidates the prompt cache.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_prompts_list_changed_notification_invalidates_cache() {
        let (mock, _notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .build();

        client.connect().await.unwrap();

        // Populate prompt cache
        {
            let mut cache = client.cached_prompts.write().await;
            *cache = Some(vec![]);
        }

        assert!(client.cached_prompts.read().await.is_some());

        event_sender
            .send(ServerEvent::Notification(json!({
                "jsonrpc": "2.0",
                "method": "notifications/prompts/list_changed"
            })))
            .unwrap();

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

        assert!(
            client.cached_prompts.read().await.is_none(),
            "Prompt cache should be invalidated after notifications/prompts/list_changed"
        );
    }

    /// Test: User notification callback is invoked on server notifications.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_user_notification_callback_fires() {
        let (mock, _notifications) = MockTransport::new();
        let event_sender = mock.event_sender();

        let received_methods = Arc::new(parking_lot::Mutex::new(Vec::<String>::new()));
        let received_methods_clone = Arc::clone(&received_methods);

        let client = McpClientBuilder::new()
            .with_transport(Box::new(mock))
            .on_notification(move |method, _params| {
                received_methods_clone.lock().push(method.to_string());
            })
            .build();

        client.connect().await.unwrap();

        // Send a tools/list_changed notification
        event_sender
            .send(ServerEvent::Notification(json!({
                "jsonrpc": "2.0",
                "method": "notifications/tools/list_changed"
            })))
            .unwrap();

        // Send a custom notification
        event_sender
            .send(ServerEvent::Notification(json!({
                "jsonrpc": "2.0",
                "method": "notifications/custom/event",
                "params": {"key": "value"}
            })))
            .unwrap();

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

        let methods = received_methods.lock();
        assert!(
            methods.contains(&"notifications/tools/list_changed".to_string()),
            "User callback should receive tools/list_changed, got: {:?}",
            *methods
        );
        assert!(
            methods.contains(&"notifications/custom/event".to_string()),
            "User callback should receive custom notifications, got: {:?}",
            *methods
        );
    }

    /// Test: refresh_tools() bypasses cache.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_refresh_tools_bypasses_cache() {
        let mut transport = StatefulMockTransport::new();

        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));

        // First list_tools
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "result": {
                "tools": [
                    {"name": "tool_a", "description": "Tool A", "inputSchema": {"type": "object"}}
                ]
            }
        })));

        // refresh_tools response (different tools)
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_2",
            "result": {
                "tools": [
                    {"name": "tool_a", "description": "Tool A", "inputSchema": {"type": "object"}},
                    {"name": "tool_b", "description": "Tool B", "inputSchema": {"type": "object"}}
                ]
            }
        })));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        let tools1 = client.list_tools().await.unwrap();
        assert_eq!(tools1.len(), 1);

        // refresh_tools should hit the server, not cache
        let tools2 = client.refresh_tools().await.unwrap();
        assert_eq!(tools2.len(), 2);

        // Subsequent list_tools should return new cache
        let tools3 = client.list_tools().await.unwrap();
        assert_eq!(tools3.len(), 2);
    }

    // ── ReconnectableMockTransport ─────────────────────────────────────
    //
    // A mock that supports multiple connect()/disconnect() cycles by
    // advertising server_events: false (skips event listener in connect()).
    // Used for testing the -32031 session retry path, which is a
    // 2025-11-25-only handshake/session concept.

    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    struct ReconnectableMockTransport {
        init_responses: Arc<std::sync::Mutex<VecDeque<McpClientResult<TransportResponse>>>>,
        request_responses: Arc<std::sync::Mutex<VecDeque<McpClientResult<Value>>>>,
        set_session_ids: Arc<std::sync::Mutex<Vec<String>>>,
        clear_count: Arc<AtomicU32>,
        connect_count: Arc<AtomicU32>,
        disconnect_count: Arc<AtomicU32>,
        connected: AtomicBool,
    }

    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    impl ReconnectableMockTransport {
        fn new() -> Self {
            Self {
                init_responses: Arc::new(std::sync::Mutex::new(VecDeque::new())),
                request_responses: Arc::new(std::sync::Mutex::new(VecDeque::new())),
                set_session_ids: Arc::new(std::sync::Mutex::new(Vec::new())),
                clear_count: Arc::new(AtomicU32::new(0)),
                connect_count: Arc::new(AtomicU32::new(0)),
                disconnect_count: Arc::new(AtomicU32::new(0)),
                connected: AtomicBool::new(false),
            }
        }

        fn push_init_response(&mut self, resp: McpClientResult<TransportResponse>) {
            self.init_responses.lock().unwrap().push_back(resp);
        }

        fn push_request_response(&mut self, resp: McpClientResult<Value>) {
            self.request_responses.lock().unwrap().push_back(resp);
        }
    }

    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[async_trait]
    impl crate::transport::Transport for ReconnectableMockTransport {
        fn transport_type(&self) -> TransportType {
            TransportType::Http
        }

        fn capabilities(&self) -> TransportCapabilities {
            TransportCapabilities {
                streaming: false,
                bidirectional: false,
                server_events: false, // No event listener — supports multiple connect cycles
                max_message_size: None,
                persistent: false,
            }
        }

        async fn connect(&self) -> McpClientResult<()> {
            self.connect_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.connected.store(true, AtomicOrdering::SeqCst);
            Ok(())
        }

        async fn disconnect(&self) -> McpClientResult<()> {
            self.disconnect_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.connected.store(false, AtomicOrdering::SeqCst);
            Ok(())
        }

        fn is_connected(&self) -> bool {
            self.connected.load(AtomicOrdering::SeqCst)
        }

        async fn send_request(&self, _request: Value) -> McpClientResult<Value> {
            self.request_responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| {
                    Err(McpClientError::generic(
                        "ReconnectableMockTransport: no more request responses queued",
                    ))
                })
        }

        async fn send_request_with_headers(
            &self,
            request: Value,
        ) -> McpClientResult<TransportResponse> {
            // Answer the version probe like a 2025-11-25 server (-32601) without
            // consuming a queued init response, so fallback drives initialize.
            if request.get("method").and_then(|m| m.as_str()) == Some("server/discover") {
                return Ok(TransportResponse::new(
                    json!({"jsonrpc": "2.0", "id": "req_0", "error": {"code": -32601, "message": "Method not found"}}),
                    HashMap::new(),
                ));
            }
            self.init_responses
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or_else(|| {
                    Err(McpClientError::generic(
                        "ReconnectableMockTransport: no more init responses queued",
                    ))
                })
        }

        async fn send_notification(&self, _notification: Value) -> McpClientResult<()> {
            Ok(())
        }

        async fn send_delete(&self, _session_id: &str) -> McpClientResult<()> {
            Ok(())
        }

        fn set_session_id(&self, session_id: String) {
            self.set_session_ids.lock().unwrap().push(session_id);
        }

        fn clear_session_id(&self) {
            self.clear_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        }

        async fn start_event_listener(&self) -> McpClientResult<EventReceiver> {
            // Should never be called since server_events is false
            Err(McpClientError::generic("No event listener"))
        }

        fn connection_info(&self) -> ConnectionInfo {
            ConnectionInfo {
                transport_type: TransportType::Http,
                endpoint: "reconnectable-mock://test".to_string(),
                connected: self.is_connected(),
                capabilities: self.capabilities(),
                metadata: Value::Null,
            }
        }

        fn statistics(&self) -> TransportStatistics {
            TransportStatistics::default()
        }
    }

    // ── Session not-initialized (-32031) retry tests ──────────────────

    /// Test: -32031 error triggers disconnect + reconnect + single retry.
    /// Simulates the race condition where notifications/initialized hasn't
    /// been processed when the first tool call arrives.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_session_not_initialized_triggers_reconnect_and_retry() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect → initialize succeeds with session "AAA"
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. First call_tool → server returns -32031 (initialized not processed yet)
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32031,
                "message": "Session error: Session not initialized - client must send notifications/initialized first (strict lifecycle mode)"
            }
        })));

        // 3. Reconnect → initialize succeeds with session "BBB"
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));

        // 4. Retried call_tool → succeeds
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_2",
            "result": {
                "content": [{"type": "text", "text": "42"}],
                "isError": false
            }
        })));

        let connect_count = transport.connect_count.clone();
        let disconnect_count = transport.disconnect_count.clone();
        let set_ids = transport.set_session_ids.clone();

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        // This should: fail with -32031 → disconnect → reconnect → retry → succeed
        let result = client.call_tool("add", json!({"a": 1, "b": 2})).await;
        assert!(
            result.is_ok(),
            "call_tool should succeed after -32031 reconnect: {:?}",
            result.err()
        );

        let call_result = result.unwrap();
        assert!(!call_result.is_error.unwrap_or(true));

        // Verify disconnect + reconnect happened
        // connect_count: 1 (initial) + 1 (reconnect) = 2
        assert_eq!(
            connect_count.load(Ordering::SeqCst),
            2,
            "Should have connected twice (initial + reconnect)"
        );
        assert_eq!(
            disconnect_count.load(Ordering::SeqCst),
            1,
            "Should have disconnected once during recovery"
        );

        // Verify new session ID was set
        let ids = set_ids.lock().unwrap();
        assert!(
            ids.contains(&"session-BBB".to_string()),
            "New session ID 'session-BBB' should be set after reconnect, got: {:?}",
            *ids
        );
    }

    /// Test: -32031 reconnect only retries once — if retry also fails,
    /// the retry error is returned (not the original -32031).
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_session_not_initialized_retry_fails_returns_retry_error() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. First request → -32031
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32031,
                "message": "Session error: Session not initialized"
            }
        })));

        // 3. Reconnect succeeds
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));

        // 4. Retry ALSO fails with -32031 (persistent issue)
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_2",
            "error": {
                "code": -32031,
                "message": "Session error: Session not initialized (still)"
            }
        })));

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        let result = client.call_tool("add", json!({})).await;
        assert!(result.is_err(), "Should fail after retry also fails");

        let err = result.unwrap_err();
        assert_eq!(
            err.error_code(),
            Some(-32031),
            "Should return the retry's error"
        );
    }

    /// Test: -32031 when reconnect itself fails — original error is returned.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_session_not_initialized_reconnect_fails_returns_original_error() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. Request → -32031
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32031,
                "message": "Session error: Session not initialized"
            }
        })));

        // 3. Reconnect FAILS (no init response queued → will error)

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        let result = client.call_tool("add", json!({})).await;
        assert!(result.is_err(), "Should fail when reconnect fails");

        // Original -32031 error is returned
        let err = result.unwrap_err();
        assert!(
            err.is_session_not_initialized(),
            "Should return original -32031 error when reconnect fails, got: {}",
            err
        );
    }

    /// Test: "Session not initialized" message without code -32031 is still detected.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_session_not_initialized_detected_by_message() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. Request → error with different code but matching message
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32000,
                "message": "Session not initialized - client must send notifications/initialized first"
            }
        })));

        // 3. Reconnect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));

        // 4. Retry succeeds
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_2",
            "result": { "tools": [] }
        })));

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        let result = client.list_tools().await;
        assert!(
            result.is_ok(),
            "Message-based detection should trigger reconnect: {:?}",
            result.err()
        );
    }

    /// Test: Non-session errors (e.g. -32602) do NOT trigger reconnect.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_non_session_error_does_not_trigger_reconnect() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. Request → -32602 (invalid params — not a session error)
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32602,
                "message": "Invalid params: missing 'name'"
            }
        })));

        let connect_count = transport.connect_count.clone();

        let client = McpClient::new(Box::new(transport), fast_retry_config(1));
        client.connect().await.unwrap();

        let result = client.call_tool("add", json!({})).await;
        assert!(result.is_err());

        // Should NOT have reconnected — only the initial connect
        assert_eq!(
            connect_count.load(Ordering::SeqCst),
            1,
            "Non-session errors must not trigger reconnect"
        );
    }

    /// Test: call_tool_with_task also benefits from -32031 retry
    /// (it uses send_request_internal internally).
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_call_tool_with_task_also_retries_on_session_error() {
        let mut transport = ReconnectableMockTransport::new();

        // 1. Initial connect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-AAA"),
            "2025-11-25",
        )));

        // 2. call_tool_with_task → -32031
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_1",
            "error": {
                "code": -32031,
                "message": "Session error: Session not initialized"
            }
        })));

        // 3. Reconnect
        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-BBB"),
            "2025-11-25",
        )));

        // 4. Retry succeeds (sync tool result)
        transport.push_request_response(Ok(json!({
            "jsonrpc": "2.0",
            "id": "req_2",
            "result": {
                "content": [{"type": "text", "text": "done"}],
                "isError": false
            }
        })));

        let client = McpClient::new(Box::new(transport), fast_retry_config(3));
        client.connect().await.unwrap();

        let result = client
            .call_tool_with_task("slow_add", json!({"a": 1}), None)
            .await;
        assert!(
            result.is_ok(),
            "call_tool_with_task should recover from -32031: {:?}",
            result.err()
        );
    }

    /// Test: invalidate_caches() clears all caches.
    #[cfg(any(feature = "client-bilingual", feature = "client-2025-11-25-only"))]
    #[tokio::test]
    async fn test_invalidate_caches_clears_all() {
        let mut transport = StatefulMockTransport::new();

        transport.push_init_response(Ok(StatefulMockTransport::make_init_response(
            Some("session-1"),
            "2025-11-25",
        )));

        let client = McpClient::new(Box::new(transport), ClientConfig::default());
        client.connect().await.unwrap();

        // Populate all caches manually
        *client.cached_tools.write().await = Some(vec![]);
        *client.cached_resources.write().await = Some(vec![]);
        *client.cached_prompts.write().await = Some(vec![]);

        client.invalidate_caches().await;

        assert!(client.cached_tools.read().await.is_none());
        assert!(client.cached_resources.read().await.is_none());
        assert!(client.cached_prompts.read().await.is_none());
    }

    // ========================================================================
    // JSON-RPC envelope construction tests (turul-rpc typed constructors)
    //
    // These tests guard the sweep that replaced 20+ hand-rolled
    // `json!({"jsonrpc": "2.0", ...})` envelopes with `JsonRpcRequest::new` /
    // `JsonRpcNotification::new` from `turul-rpc`. They lock in the semantic
    // envelope shape (jsonrpc version, method, id presence, params field
    // present-vs-absent) so a future refactor — e.g., switching to
    // `new_no_params`, which would omit `params` via `skip_serializing_if` —
    // cannot silently change the on-wire envelope contract.
    // ========================================================================

    fn fresh_client_for_envelope_tests() -> McpClient {
        let (mock, _notifications) = MockTransport::new();
        McpClient::new(Box::new(mock), ClientConfig::default())
    }

    #[test]
    fn test_value_to_request_params_null_yields_none() {
        assert!(value_to_request_params(Value::Null).is_none());
    }

    #[test]
    fn test_value_to_request_params_empty_object_yields_some_empty() {
        let result = value_to_request_params(json!({}));
        match result {
            Some(RequestParams::Object(map)) => assert!(map.is_empty()),
            other => panic!("expected Some(Object(empty)), got {:?}", other),
        }
    }

    #[test]
    fn test_value_to_request_params_object_preserves_entries() {
        let result = value_to_request_params(json!({"name": "x", "n": 42}));
        match result {
            Some(RequestParams::Object(map)) => {
                assert_eq!(map.get("name"), Some(&json!("x")));
                assert_eq!(map.get("n"), Some(&json!(42)));
            }
            other => panic!("expected Some(Object), got {:?}", other),
        }
    }

    #[test]
    fn test_value_to_request_params_array() {
        let result = value_to_request_params(json!([1, 2, 3]));
        match result {
            Some(RequestParams::Array(arr)) => {
                assert_eq!(arr, vec![json!(1), json!(2), json!(3)]);
            }
            other => panic!("expected Some(Array), got {:?}", other),
        }
    }

    #[test]
    #[should_panic(expected = "MCP client requests use object or null params")]
    fn test_value_to_request_params_scalar_panics() {
        // Defensive: no MCP client call site passes a scalar Value as params.
        // If a future call site does, surface it loudly rather than silently
        // wrapping in a positional-array RequestParams::Array.
        let _ = value_to_request_params(json!("just-a-string"));
    }

    #[tokio::test]
    async fn test_build_request_envelope_shape() {
        let client = fresh_client_for_envelope_tests();
        let envelope = client.build_request("ping", json!({}));

        assert_eq!(envelope["jsonrpc"], json!("2.0"));
        assert_eq!(envelope["method"], json!("ping"));
        assert_eq!(envelope["id"], json!("req_0"));
        assert_eq!(
            envelope["params"],
            json!({}),
            "empty Object params MUST serialize as `\"params\":{{}}` (not omitted) for wire-byte compat"
        );
    }

    #[tokio::test]
    async fn test_build_request_with_object_params() {
        let client = fresh_client_for_envelope_tests();
        let envelope = client.build_request(
            "tools/call",
            json!({"name": "calc", "arguments": {"a": 1, "b": 2}}),
        );

        assert_eq!(envelope["method"], json!("tools/call"));
        assert_eq!(envelope["params"]["name"], json!("calc"));
        assert_eq!(envelope["params"]["arguments"]["a"], json!(1));
        assert_eq!(envelope["params"]["arguments"]["b"], json!(2));
    }

    /// Regression: MCP tool arguments commonly carry array values
    /// (e.g., `compute_stats({"values": [1.0, 2.0, 3.0]})`). The sweep routes
    /// the `arguments` Object through `RequestParams::Object(HashMap<String,
    /// serde_json::Value>)`; the inner `Value`s must survive untouched so a
    /// nested `Value::Array` is still an array on the wire — not flattened,
    /// stringified, or coerced. Distinct from JSON-RPC positional params at
    /// the envelope level (see `Value::Array` branch in `value_to_request_params`).
    #[tokio::test]
    async fn test_build_request_preserves_nested_array_values_in_arguments() {
        let client = fresh_client_for_envelope_tests();
        let envelope = client.build_request(
            "tools/call",
            json!({
                "name": "compute_stats",
                "arguments": {
                    "values": [1.0, 2.0, 3.0, 4.5],
                    "tags": ["alpha", "beta"],
                    "matrix": [[1, 2], [3, 4]],
                }
            }),
        );

        assert_eq!(envelope["method"], json!("tools/call"));
        assert_eq!(envelope["params"]["name"], json!("compute_stats"));
        assert_eq!(
            envelope["params"]["arguments"]["values"],
            json!([1.0, 2.0, 3.0, 4.5]),
            "numeric array argument must survive intact"
        );
        assert_eq!(
            envelope["params"]["arguments"]["tags"],
            json!(["alpha", "beta"]),
            "string array argument must survive intact"
        );
        assert_eq!(
            envelope["params"]["arguments"]["matrix"],
            json!([[1, 2], [3, 4]]),
            "nested-array argument must survive intact"
        );
    }

    #[tokio::test]
    async fn test_build_request_id_increments_per_call() {
        let client = fresh_client_for_envelope_tests();
        let a = client.build_request("ping", json!({}));
        let b = client.build_request("ping", json!({}));
        let c = client.build_request("ping", json!({}));

        assert_eq!(a["id"], json!("req_0"));
        assert_eq!(b["id"], json!("req_1"));
        assert_eq!(c["id"], json!("req_2"));
    }

    #[tokio::test]
    async fn test_build_request_null_params_omits_field() {
        let client = fresh_client_for_envelope_tests();
        let envelope = client.build_request("ping", Value::Null);

        assert_eq!(envelope["jsonrpc"], json!("2.0"));
        assert_eq!(envelope["method"], json!("ping"));
        assert!(
            envelope.get("params").is_none(),
            "Value::Null params MUST omit the `params` field (skip_serializing_if=Option::is_none)"
        );
    }

    #[test]
    fn test_build_notification_envelope_shape() {
        let envelope = McpClient::build_notification("notifications/initialized", json!({}));

        assert_eq!(envelope["jsonrpc"], json!("2.0"));
        assert_eq!(envelope["method"], json!("notifications/initialized"));
        assert_eq!(envelope["params"], json!({}));
        assert!(
            envelope.get("id").is_none(),
            "notifications MUST NOT include an `id` field per JSON-RPC 2.0 §4.1"
        );
    }

    #[test]
    fn test_build_notification_with_object_params() {
        let envelope = McpClient::build_notification(
            "notifications/progress",
            json!({"progressToken": "tok-1", "progress": 0.5}),
        );

        assert!(envelope.get("id").is_none());
        assert_eq!(envelope["method"], json!("notifications/progress"));
        assert_eq!(envelope["params"]["progressToken"], json!("tok-1"));
        assert_eq!(envelope["params"]["progress"], json!(0.5));
    }

    #[test]
    fn test_build_notification_null_params_omits_field() {
        let envelope = McpClient::build_notification("notifications/cancelled", Value::Null);

        assert_eq!(envelope["jsonrpc"], json!("2.0"));
        assert_eq!(envelope["method"], json!("notifications/cancelled"));
        assert!(envelope.get("params").is_none());
        assert!(envelope.get("id").is_none());
    }

    /// Locks in semantic JSON-envelope equality with the prior hand-rolled
    /// `json!({"jsonrpc": "2.0", "method": ..., "id": ..., "params": {...}})`
    /// form. Compares as `serde_json::Value`, which is field-order-agnostic per
    /// JSON-RPC 2.0 (objects are unordered key/value sets); this is not a
    /// byte-equality assertion. Catches semantic drift like an omitted `params`
    /// field, missing `jsonrpc` version, or extra envelope fields.
    #[tokio::test]
    async fn test_build_request_matches_legacy_handrolled_form() {
        let client = fresh_client_for_envelope_tests();
        let typed = client.build_request("tools/list", json!({"cursor": "abc"}));

        let legacy = json!({
            "jsonrpc": "2.0",
            "method": "tools/list",
            "id": "req_0",
            "params": {"cursor": "abc"},
        });

        assert_eq!(
            typed, legacy,
            "build_request envelope must be semantically equivalent to the prior hand-rolled form"
        );
    }

    /// "The request ID MUST NOT match the ID of any other request the sender
    /// has issued and not yet received a response for" (basic §Requests).
    /// Concurrent callers on a shared client must never observe the same id —
    /// this is the actual guarantee the atomic counter provides, distinct from
    /// the sequential-call case already covered by
    /// `test_build_request_id_increments_per_call`.
    #[tokio::test]
    async fn test_concurrent_build_request_ids_are_unique() {
        let (mock, _notifications) = MockTransport::new();
        let client = Arc::new(McpClient::new(Box::new(mock), ClientConfig::default()));

        let mut handles = Vec::new();
        for _ in 0..64 {
            let client = Arc::clone(&client);
            handles.push(tokio::spawn(async move {
                client.build_request("ping", json!({}))["id"]
                    .as_str()
                    .unwrap()
                    .to_string()
            }));
        }

        let mut ids = std::collections::HashSet::new();
        for handle in handles {
            let id = handle.await.unwrap();
            assert!(
                ids.insert(id.clone()),
                "duplicate request id observed: {id}"
            );
        }
        assert_eq!(
            ids.len(),
            64,
            "every concurrent request must get a distinct id"
        );
    }
}