tower_mcp/client/

http.rs

//! HTTP client transport for remote MCP servers.
//!
//! Provides [`HttpClientTransport`] which connects to an MCP server using
//! the Streamable HTTP transport protocol (MCP spec 2025-11-25). Manages
//! session lifecycle, SSE stream for server notifications, and HTTP POST
//! for client requests.
//!
//! # Example
//!
//! ```rust,no_run
//! use tower_mcp::client::{McpClient, HttpClientTransport};
//!
//! # async fn example() -> Result<(), tower_mcp::BoxError> {
//! let transport = HttpClientTransport::new("http://localhost:3000");
//! let client = McpClient::connect(transport).await?;
//!
//! let info = client.initialize("my-client", "1.0.0").await?;
//! println!("Connected to: {}", info.server_info.name);
//! # Ok(())
//! # }
//! ```
//!
//! # Authentication
//!
//! Transport diagnostics omit request/redirect URLs, raw HTTP error bodies,
//! and arbitrary source chains, which can contain credentials. MCP JSON-RPC
//! errors are still forwarded as protocol data; [`HttpStatusDetails`] retains
//! authentication challenges and retry metadata. Treat that data as sensitive
//! when logging it yourself.
//!
//! ```rust,no_run
//! use tower_mcp::client::{McpClient, HttpClientTransport};
//!
//! # async fn example() -> Result<(), tower_mcp::BoxError> {
//! // Bearer token
//! let transport = HttpClientTransport::new("http://localhost:3000")
//!     .bearer_token("sk-your-token-here");
//!
//! // Custom API key header
//! let transport = HttpClientTransport::new("http://localhost:3000")
//!     .api_key_header("X-API-Key", "your-key");
//!
//! // Basic auth
//! let transport = HttpClientTransport::new("http://localhost:3000")
//!     .basic_auth("user", "password");
//! # Ok(())
//! # }
//! ```

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use async_trait::async_trait;
use base64::Engine;
use tokio::sync::{Notify, RwLock, mpsc};
use tokio::task::JoinHandle;

use super::transport::ClientTransport;
use crate::error::{Error, Result};
use crate::protocol::{RequestId, notifications};

const MCP_METHOD_HEADER: &str = "mcp-method";
const MCP_NAME_HEADER: &str = "mcp-name";
const MCP_PARAM_HEADER_PREFIX: &str = "mcp-param-";
const BASE64_SENTINEL_PREFIX: &str = "=?base64?";
const BASE64_SENTINEL_SUFFIX: &str = "?=";
/// Timeout for the best-effort session-termination `DELETE`, used by both
/// `close()` and the `Drop` impl.
const SESSION_DELETE_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Debug, Clone)]
struct CustomHeaderMapping {
    suffix: String,
    property_path: Vec<String>,
}

#[cfg(feature = "oauth-client")]
#[derive(Clone)]
struct ScopeEscalationRuntime {
    handler: Arc<dyn OAuthScopeEscalationHandler>,
    state: Arc<tokio::sync::Mutex<ScopeEscalationState>>,
    max_attempts: usize,
}

#[cfg(feature = "oauth-client")]
struct ScopeEscalationState {
    scopes: Vec<String>,
    revision: usize,
}

#[cfg(feature = "oauth-client")]
impl ScopeEscalationRuntime {
    fn new<P>(handler: Arc<P>, config: OAuthScopeEscalationConfig) -> Self
    where
        P: OAuthScopeEscalationHandler,
    {
        Self {
            handler,
            state: Arc::new(tokio::sync::Mutex::new(ScopeEscalationState {
                scopes: config.initial_scopes().to_vec(),
                revision: 0,
            })),
            max_attempts: config.maximum_attempts(),
        }
    }

    async fn respond_to_challenge(
        &self,
        challenge: OAuthScopeChallenge,
        resource: &str,
        operation: &str,
        attempt: usize,
        observed_revision: usize,
    ) -> std::result::Result<ScopeEscalationDecision, OAuthClientError> {
        // Serialize the complete reauthorization flow. A second operation
        // challenged by the same scope can reuse the token produced by the
        // first rather than opening a duplicate browser/headless flow.
        let mut state = self.state.lock().await;
        let previous_scopes = state.scopes.clone();
        let mut requested_scopes = previous_scopes.clone();
        for scope in &challenge.required_scopes {
            if !requested_scopes.contains(scope) {
                requested_scopes.push(scope.clone());
            }
        }

        if requested_scopes == previous_scopes && state.revision > observed_revision {
            return Ok(ScopeEscalationDecision {
                revision: state.revision,
            });
        }
        // The scope may have been requested previously without being granted,
        // or the token may otherwise be stale. Reauthorize the same union
        // again, but only within the caller's hard attempt limit.

        self.handler
            .reauthorize(OAuthScopeEscalationRequest {
                resource: resource.to_string(),
                operation: operation.to_string(),
                challenge,
                previous_scopes,
                requested_scopes: requested_scopes.clone(),
                attempt,
            })
            .await?;

        state.scopes = requested_scopes;
        state.revision += 1;
        Ok(ScopeEscalationDecision {
            revision: state.revision,
        })
    }
}

#[cfg(feature = "oauth-client")]
struct ScopeEscalationDecision {
    revision: usize,
}

#[cfg(feature = "oauth-client")]
use super::oauth::{
    OAuthBearerChallenge, OAuthClientError, OAuthScopeChallenge, OAuthScopeEscalationConfig,
    OAuthScopeEscalationHandler, OAuthScopeEscalationRequest, TokenProvider,
};

/// Configuration for [`HttpClientTransport`].
///
/// # Example
///
/// ```rust,no_run
/// use tower_mcp::client::{HttpClientTransport, HttpClientConfig};
/// use std::time::Duration;
///
/// let config = HttpClientConfig {
///     request_timeout: Duration::from_secs(60),
///     ..Default::default()
/// };
/// let transport = HttpClientTransport::with_config("http://localhost:3000", config);
/// ```
#[derive(Debug, Clone)]
pub struct HttpClientConfig {
    /// Custom headers to include on every request (e.g., auth tokens).
    pub headers: HashMap<String, String>,
    /// Whether to automatically open the standalone SSE notification stream
    /// after initialization. Associated POST response streams used for
    /// sampling, elicitation, and roots requests work independently.
    /// Default: `true`.
    pub auto_sse: bool,
    /// Capacity of the internal message channel.
    /// Default: 256.
    pub channel_capacity: usize,
    /// Timeout for HTTP requests.
    /// Default: 30 seconds.
    pub request_timeout: Duration,
    /// Timeout for notification POSTs (frames without an `id`), capped at
    /// `request_timeout`.
    ///
    /// Notifications are awaited inline so `notifications/initialized` is
    /// ordered ahead of the first request, which blocks the client's message
    /// loop for the duration. This bounds that block independently of
    /// `request_timeout` so a server that stalls a notification's `202` cannot
    /// freeze the client for the full request timeout.
    /// Default: 5 seconds.
    pub notification_timeout: Duration,
    /// Whether to attempt SSE reconnection on disconnect.
    /// Default: `true`.
    pub sse_reconnect: bool,
    /// Delay before SSE reconnection attempts.
    /// Default: 1 second.
    pub sse_reconnect_delay: Duration,
    /// Maximum SSE reconnection attempts before giving up.
    /// Default: 5.
    pub max_sse_reconnect_attempts: u32,
    /// Whether to support automatic session recovery on expiry.
    /// When enabled, HTTP 404 responses (with a session ID attached) and
    /// JSON-RPC -32005 errors trigger re-initialization. A 404 counts
    /// whatever its body holds.
    /// Default: `true`.
    pub session_recovery: bool,
    /// Maximum size in bytes buffered for a single SSE event.
    ///
    /// A server that streams an event without ever terminating it would
    /// otherwise grow the parse buffer without bound. When a single
    /// event's buffered size exceeds this cap, the stream is terminated
    /// with [`Error::SseEventTooLarge`].
    /// Default: 16 MiB (matching rmcp).
    pub max_sse_event_size: usize,
    /// Maximum size in bytes buffered for a single non-SSE response body: a
    /// JSON-RPC response on the request path, or an error body.
    ///
    /// A server that answers with a JSON body that never ends (or is simply
    /// very large) would otherwise be buffered into memory without bound
    /// until the request timeout fires. This is checked against
    /// `Content-Length` up front and while streaming, so a response over the
    /// cap fails that request with a transport error instead; the
    /// connection itself is left open. Error bodies (used only in messages)
    /// are truncated rather than failed, independent of this setting.
    /// Default: 16 MiB (matching `max_sse_event_size`).
    pub max_response_body_size: usize,
    /// Whether to follow a redirect to a different origin (scheme, host, or
    /// port) than the one the request was sent to.
    ///
    /// By default the client follows only same-origin redirects, so a
    /// same-host `307`/`308` still works, but a server (or an attacker who
    /// controls a `Location` response) cannot redirect a request to another
    /// host and have this client forward it there. This matters because
    /// every header on the request, including custom headers set via
    /// [`HttpClientConfig::headers`] (`bearer_token`, `api_key_header`,
    /// `header`) and `mcp-session-id`, is resent to the redirect target;
    /// reqwest's own stripping only covers `Authorization`, `Cookie`, and
    /// `Proxy-Authorization`. Set this to `true` only when the server is
    /// known to redirect across origins and the forwarded headers are safe
    /// to disclose there. A redirect from `https` to a non-`https` scheme is
    /// never followed, even with this enabled, since that would send those
    /// headers in plaintext.
    ///
    /// This only widens where a redirect may lead. Whatever the origin, the
    /// client follows only `307` and `308`: they resend the request with its
    /// method and body, whereas `301`, `302`, and `303` would turn the
    /// JSON-RPC `POST` into a bodyless `GET`. Those statuses are not followed;
    /// the request fails with the 3xx status, so point the client at the
    /// final URL.
    /// Default: `false`.
    pub follow_cross_origin_redirects: bool,
}

/// Default maximum buffered size for a single SSE event (16 MiB, matching
/// rmcp). See [`HttpClientConfig::max_sse_event_size`].
pub const DEFAULT_MAX_SSE_EVENT_SIZE: usize = 16 * 1024 * 1024;

/// Default maximum buffered size for a single non-SSE response body.
/// Matches [`DEFAULT_MAX_SSE_EVENT_SIZE`]. See
/// [`HttpClientConfig::max_response_body_size`].
pub const DEFAULT_MAX_RESPONSE_BODY_SIZE: usize = DEFAULT_MAX_SSE_EVENT_SIZE;

impl Default for HttpClientConfig {
    fn default() -> Self {
        Self {
            headers: HashMap::new(),
            auto_sse: true,
            channel_capacity: 256,
            request_timeout: Duration::from_secs(30),
            notification_timeout: Duration::from_secs(5),
            sse_reconnect: true,
            sse_reconnect_delay: Duration::from_secs(1),
            max_sse_reconnect_attempts: 5,
            session_recovery: true,
            max_sse_event_size: DEFAULT_MAX_SSE_EVENT_SIZE,
            max_response_body_size: DEFAULT_MAX_RESPONSE_BODY_SIZE,
            follow_cross_origin_redirects: false,
        }
    }
}

impl HttpClientConfig {
    /// Set a Bearer token for authentication.
    pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
        self.headers.insert(
            "Authorization".to_string(),
            format!("Bearer {}", token.into()),
        );
        self
    }

    /// Set an API key using a custom header name.
    pub fn api_key_header(mut self, name: impl Into<String>, key: impl Into<String>) -> Self {
        self.headers.insert(name.into(), key.into());
        self
    }

    /// Set Basic authentication credentials.
    pub fn basic_auth(mut self, username: impl AsRef<str>, password: impl AsRef<str>) -> Self {
        use base64::Engine;
        let encoded = base64::engine::general_purpose::STANDARD.encode(format!(
            "{}:{}",
            username.as_ref(),
            password.as_ref()
        ));
        self.headers
            .insert("Authorization".to_string(), format!("Basic {}", encoded));
        self
    }

    /// Add a custom header.
    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.insert(name.into(), value.into());
        self
    }
}

/// Maximum redirect hops the custom redirect policy follows before failing,
/// matching reqwest's own default (see [`reqwest::redirect::Policy`]).
/// `Policy::custom` does not get that bound for free; it has to be checked
/// explicitly.
const MAX_REDIRECT_HOPS: usize = 10;

/// Cap on [`HttpClientTransport::pending_tools_lists`], so requests that never
/// get a response cannot grow it without bound. The oldest entry is dropped
/// first, and a response with no entry is treated as a first page.
const MAX_PENDING_TOOLS_LISTS: usize = 64;

/// Whether a redirect with this status is safe to follow for an MCP request.
///
/// Only `307` and `308` resend the request with its method and body. For
/// `301`, `302`, and `303` reqwest rewrites a `POST` into a bodyless `GET`, so
/// the JSON-RPC message would be dropped and the server would answer a request
/// it never received. The redirect policy cannot see the method, so this
/// applies to the SSE `GET` and the session `DELETE` too; the `initialize`
/// `POST` has to succeed before either is sent.
fn redirect_preserves_request(status: reqwest::StatusCode) -> bool {
    matches!(
        status,
        reqwest::StatusCode::TEMPORARY_REDIRECT | reqwest::StatusCode::PERMANENT_REDIRECT
    )
}

/// Whether a redirect from `original` (the request's starting URL) to
/// `target` (the `Location` it was just pointed at) should be followed.
///
/// Same-origin means matching scheme, host, and port, with each scheme's
/// default port substituted when a URL omits one. A downgrade from `https`
/// to any other scheme is refused unconditionally, since it would put
/// whatever headers are on the request on the wire in plaintext; that check
/// runs even when `allow_cross_origin` opts into following other origins.
///
/// Factored out of the `reqwest::redirect::Policy::custom` closure so it can
/// be unit-tested without standing up TLS.
fn should_follow_redirect(
    original: &reqwest::Url,
    target: &reqwest::Url,
    allow_cross_origin: bool,
) -> bool {
    if original.scheme() == "https" && target.scheme() != "https" {
        return false;
    }
    if allow_cross_origin {
        return true;
    }
    original.scheme() == target.scheme()
        && original.host_str() == target.host_str()
        && original.port_or_known_default() == target.port_or_known_default()
}

/// Build the `reqwest::Client` used for MCP requests, with a redirect policy
/// that keeps credentials (custom headers, `mcp-session-id`) from leaking to
/// a redirect target reqwest's own header stripping does not cover. See
/// [`HttpClientConfig::follow_cross_origin_redirects`].
fn build_client(allow_cross_origin_redirects: bool) -> reqwest::Client {
    let policy = reqwest::redirect::Policy::custom(move |attempt| {
        // Hand the 3xx response back instead of following it: the caller then
        // gets a plain non-success status (see `http_status_error`), and the
        // transport stays connected, unlike a policy error.
        if !redirect_preserves_request(attempt.status()) {
            return attempt.stop();
        }
        if attempt.previous().len() > MAX_REDIRECT_HOPS {
            return attempt.error("too many redirects");
        }
        // `previous()` holds every URL requested so far in this chain; the
        // first entry is the original request, never a redirect target.
        let Some(original) = attempt.previous().first() else {
            return attempt.follow();
        };
        if should_follow_redirect(original, attempt.url(), allow_cross_origin_redirects) {
            attempt.follow()
        } else {
            attempt.error(
                "redirect to a different origin was not followed; set \
                 HttpClientConfig::follow_cross_origin_redirects to allow it (an https-to-http \
                 downgrade is never followed)",
            )
        }
    });
    reqwest::Client::builder()
        .redirect(policy)
        .build()
        .expect("reqwest client with a custom redirect policy should always build")
}

/// Client transport for MCP servers over Streamable HTTP.
///
/// Connects to a remote MCP server using the Streamable HTTP transport
/// protocol. Manages session lifecycle (`mcp-session-id`), opens an SSE
/// stream for server-initiated messages, and sends client requests via
/// HTTP POST.
///
/// Transport-generated diagnostics omit URLs, raw HTTP error bodies, and
/// untrusted error details. They retain error categories and HTTP status codes.
/// MCP JSON-RPC errors and [`HttpStatusDetails`] remain protocol data and may
/// contain sensitive values; avoid logging them verbatim.
///
/// # How it works
///
/// The transport bridges HTTP's request/response model with the
/// `ClientTransport` trait's `send()`/`recv()` message-passing model:
///
/// - **`send()`** POSTs JSON-RPC messages to the server and queues the
///   response body into an internal channel for `recv()` to return.
/// - **`recv()`** reads from that channel, which also receives SSE events
///   from a background task.
///
/// After the `initialize` handshake establishes a session, an SSE stream
/// is automatically opened to receive server notifications and
/// server-initiated requests.
///
/// # Example
///
/// ```rust,no_run
/// use tower_mcp::client::{McpClient, HttpClientTransport};
///
/// # async fn example() -> Result<(), tower_mcp::BoxError> {
/// let transport = HttpClientTransport::new("http://localhost:3000");
/// let client = McpClient::connect(transport).await?;
///
/// let info = client.initialize("my-client", "1.0.0").await?;
/// let tools = client.list_tools().await?;
/// client.shutdown().await?;
/// # Ok(())
/// # }
/// ```
pub struct HttpClientTransport {
    /// The base URL of the MCP server endpoint.
    url: String,
    /// reqwest HTTP client (reused across requests).
    client: reqwest::Client,
    /// Session ID received from the server after `initialize`.
    session_id: Option<String>,
    /// Negotiated protocol version.
    protocol_version: Option<String>,
    /// Validated custom-header mappings learned from `tools/list`. Replaced
    /// by a first-page response and extended by later pages.
    tool_header_mappings: HashMap<String, Vec<CustomHeaderMapping>>,
    /// Outbound `tools/list` request ids on the 2026-07-28 path, each with
    /// whether it carried a `cursor`. The response frame names only the id, so
    /// this is how [`normalize_incoming_message`](Self::normalize_incoming_message)
    /// tells a later page from the first. Bounded by [`MAX_PENDING_TOOLS_LISTS`].
    pending_tools_lists: Vec<(serde_json::Value, bool)>,
    /// Channel receiver for incoming messages (POST responses + SSE events).
    incoming_rx: mpsc::Receiver<String>,
    /// Channel sender used by `send()` to queue POST response bodies
    /// and cloned for the SSE background task.
    incoming_tx: mpsc::Sender<String>,
    /// Handle to the SSE background task, if running.
    sse_task: Option<JoinHandle<()>>,
    /// In-flight POST response streams, keyed by their JSON-RPC request ID.
    ///
    /// Final `subscriptions/listen` requests stay here until cancelled or the
    /// server closes them. Other completed tasks are pruned on subsequent
    /// sends.
    request_tasks: HashMap<RequestId, JoinHandle<()>>,
    /// The id of the `initialize` request, which the spec forbids cancelling.
    initialize_request_id: Option<RequestId>,
    /// The last SSE event ID received, for stream resumption.
    last_event_id: Arc<RwLock<Option<String>>>,
    /// Server-requested retry delay from SSE `retry:` field.
    sse_retry_delay: Arc<RwLock<Option<Duration>>>,
    /// Signal to tell the SSE loop to close its current stream and reconnect.
    sse_reconnect_signal: Arc<Notify>,
    /// Whether the transport is still connected.
    connected: Arc<AtomicBool>,
    /// Configuration options.
    config: HttpClientConfig,
    /// Dynamic token provider for OAuth or other token-based auth.
    #[cfg(feature = "oauth-client")]
    token_provider: Option<Arc<dyn TokenProvider>>,
    /// Runtime policy for insufficient-scope challenges.
    #[cfg(feature = "oauth-client")]
    scope_escalation: Option<ScopeEscalationRuntime>,
}

impl HttpClientTransport {
    /// Create a new HTTP client transport targeting the given URL.
    ///
    /// Uses default configuration. The URL should be the MCP server's
    /// Streamable HTTP endpoint (e.g., `http://localhost:3000`).
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000");
    /// ```
    pub fn new(url: impl Into<String>) -> Self {
        Self::with_config(url, HttpClientConfig::default())
    }

    /// Create with custom configuration.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::{HttpClientTransport, HttpClientConfig};
    /// use std::time::Duration;
    ///
    /// let config = HttpClientConfig {
    ///     request_timeout: Duration::from_secs(60),
    ///     sse_reconnect: false,
    ///     ..Default::default()
    /// };
    /// let transport = HttpClientTransport::with_config("http://localhost:3000", config);
    /// ```
    pub fn with_config(url: impl Into<String>, config: HttpClientConfig) -> Self {
        let (tx, rx) = mpsc::channel(config.channel_capacity);
        Self {
            url: url.into(),
            client: build_client(config.follow_cross_origin_redirects),
            session_id: None,
            protocol_version: None,
            tool_header_mappings: HashMap::new(),
            pending_tools_lists: Vec::new(),
            incoming_rx: rx,
            incoming_tx: tx,
            sse_task: None,
            request_tasks: HashMap::new(),
            initialize_request_id: None,
            last_event_id: Arc::new(RwLock::new(None)),
            sse_retry_delay: Arc::new(RwLock::new(None)),
            sse_reconnect_signal: Arc::new(Notify::new()),
            connected: Arc::new(AtomicBool::new(true)),
            config,
            #[cfg(feature = "oauth-client")]
            token_provider: None,
            #[cfg(feature = "oauth-client")]
            scope_escalation: None,
        }
    }

    /// Create with an existing `reqwest::Client`.
    ///
    /// Use this when you need custom TLS configuration, proxy settings,
    /// or connection pooling.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let client = reqwest::Client::builder()
    ///     .danger_accept_invalid_certs(true) // for development
    ///     .build()
    ///     .unwrap();
    /// let transport = HttpClientTransport::with_client("https://mcp.example.com", client);
    /// ```
    pub fn with_client(url: impl Into<String>, client: reqwest::Client) -> Self {
        let config = HttpClientConfig::default();
        let (tx, rx) = mpsc::channel(config.channel_capacity);
        Self {
            url: url.into(),
            client,
            session_id: None,
            protocol_version: None,
            tool_header_mappings: HashMap::new(),
            pending_tools_lists: Vec::new(),
            incoming_rx: rx,
            incoming_tx: tx,
            sse_task: None,
            request_tasks: HashMap::new(),
            initialize_request_id: None,
            last_event_id: Arc::new(RwLock::new(None)),
            sse_retry_delay: Arc::new(RwLock::new(None)),
            sse_reconnect_signal: Arc::new(Notify::new()),
            connected: Arc::new(AtomicBool::new(true)),
            config,
            #[cfg(feature = "oauth-client")]
            token_provider: None,
            #[cfg(feature = "oauth-client")]
            scope_escalation: None,
        }
    }

    /// Set a Bearer token for `Authorization: Bearer <token>` authentication.
    ///
    /// The token is included on every HTTP request (POST and SSE GET).
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .bearer_token("sk-my-secret-token");
    /// ```
    pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
        self.config.headers.insert(
            "Authorization".to_string(),
            format!("Bearer {}", token.into()),
        );
        self
    }

    /// Set an API key for authentication.
    ///
    /// Sends as `Authorization: Bearer <key>`. Use
    /// [`api_key_header`](Self::api_key_header) for a custom header name.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .api_key("sk-my-api-key");
    /// ```
    pub fn api_key(self, key: impl Into<String>) -> Self {
        self.bearer_token(key)
    }

    /// Set an API key using a custom header name.
    ///
    /// Sends the key as the raw header value (no `Bearer` prefix).
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .api_key_header("X-API-Key", "sk-my-api-key");
    /// ```
    pub fn api_key_header(mut self, name: impl Into<String>, key: impl Into<String>) -> Self {
        self.config.headers.insert(name.into(), key.into());
        self
    }

    /// Set Basic authentication credentials.
    ///
    /// Encodes `username:password` as Base64 and sends as
    /// `Authorization: Basic <encoded>`.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .basic_auth("admin", "secret");
    /// ```
    pub fn basic_auth(mut self, username: impl AsRef<str>, password: impl AsRef<str>) -> Self {
        use base64::Engine;
        let encoded = base64::engine::general_purpose::STANDARD.encode(format!(
            "{}:{}",
            username.as_ref(),
            password.as_ref()
        ));
        self.config
            .headers
            .insert("Authorization".to_string(), format!("Basic {}", encoded));
        self
    }

    /// Add a custom header to every request.
    ///
    /// Can be called multiple times to add multiple headers.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::HttpClientTransport;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .header("X-Custom-Header", "my-value")
    ///     .header("X-Request-Source", "my-app");
    /// ```
    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.config.headers.insert(name.into(), value.into());
        self
    }

    /// Disable automatic session recovery.
    ///
    /// By default, if the server returns a session expired error (HTTP 404
    /// with session ID or JSON-RPC -32005), the client will automatically
    /// re-initialize and retry the failed operation. Call this to disable
    /// that behavior and surface the error to the caller instead.
    pub fn disable_session_recovery(mut self) -> Self {
        self.config.session_recovery = false;
        self
    }

    /// Set a dynamic token provider for authentication.
    ///
    /// The provider's [`TokenProvider::get_token()`] is called before each
    /// HTTP request, and the returned token is sent as `Authorization: Bearer <token>`.
    /// This overrides any static `Authorization` header set via [`bearer_token()`](Self::bearer_token)
    /// or [`basic_auth()`](Self::basic_auth).
    ///
    /// Use [`OAuthClientCredentials`](super::OAuthClientCredentials) for
    /// OAuth 2.0 Client Credentials grants, or implement [`TokenProvider`]
    /// for custom token acquisition logic.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use tower_mcp::client::{HttpClientTransport, OAuthClientCredentials};
    ///
    /// # fn example() -> Result<(), tower_mcp::BoxError> {
    /// let provider = OAuthClientCredentials::builder()
    ///     .client_id("my-client")
    ///     .client_secret("my-secret")
    ///     .token_endpoint("https://auth.example.com/token")
    ///     .resource("http://localhost:3000")
    ///     .build()?;
    ///
    /// let transport = HttpClientTransport::new("http://localhost:3000")
    ///     .with_token_provider(provider);
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "oauth-client")]
    pub fn with_token_provider(mut self, provider: impl TokenProvider) -> Self {
        self.token_provider = Some(Arc::new(provider));
        self.scope_escalation = None;
        self
    }

    /// Set a token provider with bounded runtime scope escalation.
    ///
    /// When an MCP operation receives an HTTP 403 Bearer challenge with
    /// `error="insufficient_scope"`, the transport unions the challenged
    /// scopes with the scopes already tracked by `config`, invokes
    /// [`OAuthScopeEscalationHandler::reauthorize`], asks the provider for a
    /// fresh token, and retries the same operation. Reauthorization is
    /// serialized across concurrent requests, and each operation is bounded
    /// by [`OAuthScopeEscalationConfig::maximum_attempts`].
    ///
    /// The provider and handler are the same value so the handler can update
    /// the token returned by [`TokenProvider::get_token`].
    #[cfg(feature = "oauth-client")]
    pub fn with_scope_aware_token_provider<P>(
        mut self,
        provider: P,
        config: OAuthScopeEscalationConfig,
    ) -> Self
    where
        P: TokenProvider + OAuthScopeEscalationHandler,
    {
        let provider = Arc::new(provider);
        self.token_provider = Some(provider.clone());
        self.scope_escalation = Some(ScopeEscalationRuntime::new(provider, config));
        self
    }

    fn outgoing_custom_headers(&self, parsed: &serde_json::Value) -> Vec<(String, String)> {
        if parsed.get("method").and_then(serde_json::Value::as_str) != Some("tools/call") {
            return Vec::new();
        }
        let Some(params) = parsed.get("params") else {
            return Vec::new();
        };
        let Some(name) = params.get("name").and_then(serde_json::Value::as_str) else {
            return Vec::new();
        };
        let Some(mappings) = self.tool_header_mappings.get(name) else {
            return Vec::new();
        };
        let arguments = params.get("arguments").unwrap_or(&serde_json::Value::Null);

        mappings
            .iter()
            .filter_map(|mapping| {
                let value = value_at_property_path(arguments, &mapping.property_path)?;
                if value.is_null() {
                    return None;
                }
                let rendered = json_value_to_header_string(value)?;
                Some((
                    format!("{MCP_PARAM_HEADER_PREFIX}{}", mapping.suffix),
                    encode_header_value(&rendered),
                ))
            })
            .collect()
    }

    fn normalize_incoming_message(&mut self, message: String) -> String {
        if self.protocol_version.as_deref() != Some(crate::protocol::PROTOCOL_VERSION_2026_07_28) {
            return message;
        }
        let Ok(mut parsed) = serde_json::from_str::<serde_json::Value>(&message) else {
            return message;
        };
        // The server says the tool set changed, so every learned mapping is
        // suspect. The client re-lists from the first page next.
        if parsed.get("method").and_then(serde_json::Value::as_str)
            == Some("notifications/tools/list_changed")
        {
            self.tool_header_mappings.clear();
            return message;
        }
        // Take the recorded outbound request, if any, even for a frame that
        // carries no tools (an error), so the entry does not linger.
        let later_page = parsed
            .get("id")
            .and_then(|id| {
                let index = self
                    .pending_tools_lists
                    .iter()
                    .position(|(pending, _)| json_request_ids_match(pending, id))?;
                Some(self.pending_tools_lists.remove(index).1)
            })
            .unwrap_or(false);
        let Some(tools) = parsed
            .get_mut("result")
            .and_then(|result| result.get_mut("tools"))
            .and_then(serde_json::Value::as_array_mut)
        else {
            return message;
        };

        // `list_all_tools` pages, and a later page must add to the mappings
        // of the earlier ones rather than replace them: a call to a tool from
        // page one still has to mirror its `Mcp-Param-*` headers.
        if !later_page {
            self.tool_header_mappings.clear();
        }
        tools.retain(|tool| {
            let Some(name) = tool.get("name").and_then(serde_json::Value::as_str) else {
                return false;
            };
            let Some(schema) = tool.get("inputSchema") else {
                return false;
            };
            match custom_header_mappings(schema) {
                Ok(mappings) => {
                    self.tool_header_mappings.insert(name.to_string(), mappings);
                    true
                }
                Err(error) => {
                    tracing::warn!(tool = %name, %error, "Excluding tool with invalid x-mcp-header annotations");
                    false
                }
            }
        });

        parsed.to_string()
    }

    /// Queue a frame for [`recv`](ClientTransport::recv).
    async fn queue_incoming(&self, frame: String) -> Result<()> {
        self.incoming_tx
            .send(frame)
            .await
            .map_err(|_| Error::Transport("Internal channel closed".to_string()))
    }

    /// Start the SSE background stream after session is established.
    fn start_sse_stream(&mut self) {
        let url = self.url.clone();
        let client = self.client.clone();
        let session_id = self.session_id.clone().unwrap();
        let protocol_version = self.protocol_version.clone();
        let tx = self.incoming_tx.clone();
        let last_event_id = self.last_event_id.clone();
        let sse_retry_delay = self.sse_retry_delay.clone();
        let reconnect_signal = self.sse_reconnect_signal.clone();
        let connected = self.connected.clone();
        let config = self.config.clone();
        #[cfg(feature = "oauth-client")]
        let token_provider = self.token_provider.clone();

        self.sse_task = Some(tokio::spawn(async move {
            sse_stream_loop(SseLoopParams {
                url,
                client,
                session_id,
                protocol_version,
                tx,
                last_event_id,
                sse_retry_delay,
                reconnect_signal,
                connected,
                config,
                #[cfg(feature = "oauth-client")]
                token_provider,
            })
            .await;
        }));
    }
}

fn custom_header_mappings(
    schema: &serde_json::Value,
) -> std::result::Result<Vec<CustomHeaderMapping>, String> {
    fn annotation_count(value: &serde_json::Value) -> usize {
        match value {
            serde_json::Value::Object(object) => {
                usize::from(object.contains_key("x-mcp-header"))
                    + object.values().map(annotation_count).sum::<usize>()
            }
            serde_json::Value::Array(values) => values.iter().map(annotation_count).sum::<usize>(),
            _ => 0,
        }
    }

    fn is_tchar(byte: u8) -> bool {
        byte.is_ascii_alphanumeric()
            || matches!(
                byte,
                b'!' | b'#'
                    | b'$'
                    | b'%'
                    | b'&'
                    | b'\''
                    | b'*'
                    | b'+'
                    | b'-'
                    | b'.'
                    | b'^'
                    | b'_'
                    | b'`'
                    | b'|'
                    | b'~'
            )
    }

    // SEP-2243 allows integer, string and boolean; `number` is not permitted.
    fn primitive_header_type(schema: &serde_json::Value) -> bool {
        match schema.get("type") {
            Some(serde_json::Value::String(kind)) => {
                matches!(kind.as_str(), "string" | "integer" | "boolean")
            }
            Some(serde_json::Value::Array(kinds)) => {
                let mut primitive = false;
                for kind in kinds {
                    match kind.as_str() {
                        Some("string" | "integer" | "boolean") if !primitive => {
                            primitive = true;
                        }
                        Some("null") => {}
                        _ => return false,
                    }
                }
                primitive
            }
            _ => false,
        }
    }

    fn walk(
        schema: &serde_json::Value,
        path: &mut Vec<String>,
        seen: &mut std::collections::HashSet<String>,
        mappings: &mut Vec<CustomHeaderMapping>,
    ) -> std::result::Result<(), String> {
        let Some(properties) = schema
            .get("properties")
            .and_then(serde_json::Value::as_object)
        else {
            return Ok(());
        };
        for (property_name, property_schema) in properties {
            path.push(property_name.clone());
            if let Some(annotation) = property_schema.get("x-mcp-header") {
                let suffix = annotation
                    .as_str()
                    .ok_or_else(|| format!("annotation at {} is not a string", path.join(".")))?;
                if suffix.is_empty() || !suffix.bytes().all(is_tchar) {
                    return Err(format!(
                        "invalid header suffix {suffix:?} at {}",
                        path.join(".")
                    ));
                }
                if !primitive_header_type(property_schema) {
                    return Err(format!(
                        "annotation at {} is not on a primitive property",
                        path.join(".")
                    ));
                }
                if !seen.insert(suffix.to_ascii_lowercase()) {
                    return Err(format!("duplicate header suffix {suffix:?}"));
                }
                mappings.push(CustomHeaderMapping {
                    suffix: suffix.to_string(),
                    property_path: path.clone(),
                });
            }
            walk(property_schema, path, seen, mappings)?;
            path.pop();
        }
        Ok(())
    }

    let mut mappings = Vec::new();
    walk(
        schema,
        &mut Vec::new(),
        &mut std::collections::HashSet::new(),
        &mut mappings,
    )?;
    if annotation_count(schema) != mappings.len() {
        return Err(
            "x-mcp-header annotation is not statically reachable through properties".to_string(),
        );
    }
    Ok(mappings)
}

fn value_at_property_path<'a>(
    root: &'a serde_json::Value,
    path: &[String],
) -> Option<&'a serde_json::Value> {
    path.iter().try_fold(root, |value, key| value.get(key))
}

fn json_value_to_header_string(value: &serde_json::Value) -> Option<String> {
    match value {
        serde_json::Value::String(value) => Some(value.clone()),
        serde_json::Value::Number(value) => Some(value.to_string()),
        serde_json::Value::Bool(value) => Some(value.to_string()),
        serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => {
            None
        }
    }
}

fn encode_header_value(value: &str) -> String {
    let unsafe_for_header =
        value.trim() != value || value.bytes().any(|byte| !(0x20..=0x7e).contains(&byte));
    if unsafe_for_header {
        format!(
            "{BASE64_SENTINEL_PREFIX}{}{BASE64_SENTINEL_SUFFIX}",
            base64::engine::general_purpose::STANDARD.encode(value)
        )
    } else {
        value.to_string()
    }
}

#[cfg(feature = "oauth-client")]
fn bearer_headers(token: &str) -> std::result::Result<reqwest::header::HeaderMap, String> {
    let value = reqwest::header::HeaderValue::from_str(&format!("Bearer {token}"))
        .map_err(|_| "token provider returned an invalid bearer token".to_string())?;
    let mut headers = reqwest::header::HeaderMap::new();
    headers.insert(reqwest::header::AUTHORIZATION, value);
    // RequestBuilder::header appends, which can leave a stale Authorization
    // value in front of the fresh token. Supplying a HeaderMap replaces the
    // existing value instead.
    Ok(headers)
}

fn is_jsonrpc_error_response(value: &serde_json::Value) -> bool {
    value.get("error").is_some_and(serde_json::Value::is_object)
        && value.pointer("/error/code").is_some()
        && value.pointer("/error/message").is_some()
}

struct HttpRequestSendError {
    message: String,
    connection_failed: bool,
}

// Neither Display, Debug, nor a source chain is safe: reqwest carries the URL,
// and a custom redirect error can quote an entirely different URL.
fn request_error_message(error: &reqwest::Error) -> String {
    let category = if error.is_timeout() {
        "timeout"
    } else if error.is_redirect() {
        "redirect rejected (check redirect policy)"
    } else if error.is_connect() {
        "connection failed"
    } else if error.is_builder() {
        "invalid request configuration"
    } else if error.is_body() {
        "response body I/O failed"
    } else if error.is_decode() {
        "response decoding failed"
    } else {
        "request failed"
    };
    match error.status() {
        Some(status) => format!("HTTP request failed: {category} (HTTP {status})"),
        None => format!("HTTP request failed: {category}"),
    }
}

impl HttpRequestSendError {
    fn request(error: reqwest::Error) -> Self {
        Self {
            message: request_error_message(&error),
            connection_failed: true,
        }
    }

    #[cfg(feature = "oauth-client")]
    fn oauth(_error: OAuthClientError) -> Self {
        Self {
            message: "OAuth token acquisition failed".to_string(),
            connection_failed: false,
        }
    }
}

#[cfg(not(feature = "oauth-client"))]
async fn send_http_request(
    request: reqwest::RequestBuilder,
    resource: &str,
    operation: &str,
) -> std::result::Result<reqwest::Response, HttpRequestSendError> {
    let _ = (resource, operation);
    request.send().await.map_err(HttpRequestSendError::request)
}

#[cfg(feature = "oauth-client")]
async fn send_http_request(
    mut request: reqwest::RequestBuilder,
    resource: &str,
    operation: &str,
    token_provider: Option<Arc<dyn TokenProvider>>,
    scope_escalation: Option<ScopeEscalationRuntime>,
    initial_scope_revision: usize,
) -> std::result::Result<reqwest::Response, HttpRequestSendError> {
    let mut observed_revision = initial_scope_revision;
    let mut attempts = 0;
    // Bounded separately from scope escalation, and at one. The two are
    // different failures: a scope challenge names what to ask for and can
    // legitimately step up more than once, while a rejected token has exactly
    // one remedy, and repeating it against a server that keeps saying no is a
    // retry loop rather than a recovery (#1370).
    let mut reauthenticated = false;

    loop {
        let retry_request = request.try_clone();

        let response = request
            .send()
            .await
            .map_err(HttpRequestSendError::request)?;

        // A 401 says the token was rejected whatever the provider's cache
        // believes, so the cached value is discarded before asking for
        // another. Anything without a Bearer challenge is left alone: a bare
        // 401 is the caller's to handle.
        if response.status() == reqwest::StatusCode::UNAUTHORIZED
            && !reauthenticated
            && bearer_challenge(response.headers()).is_some()
            && let Some(provider) = token_provider.as_ref()
            // Cloned rather than taken: the scope-escalation path below needs
            // the original if this branch does not fire.
            && let Some(reauth_request) = retry_request.as_ref().and_then(|r| r.try_clone())
        {
            reauthenticated = true;
            provider.invalidate().await;
            let token = provider
                .get_token()
                .await
                .map_err(HttpRequestSendError::oauth)?;
            let headers = bearer_headers(&token).map_err(|message| {
                HttpRequestSendError::oauth(OAuthClientError::ScopeEscalation(message))
            })?;
            request = reauth_request.headers(headers);
            continue;
        }

        let challenge = if response.status() == reqwest::StatusCode::FORBIDDEN {
            scope_challenge(response.headers())
        } else {
            None
        };
        let Some(challenge) = challenge else {
            return Ok(response);
        };
        let (Some(runtime), Some(provider), Some(mut retry_request)) = (
            scope_escalation.as_ref(),
            token_provider.as_ref(),
            retry_request,
        ) else {
            return Ok(response);
        };
        if attempts >= runtime.max_attempts {
            return Ok(response);
        }

        attempts += 1;
        let decision = runtime
            .respond_to_challenge(challenge, resource, operation, attempts, observed_revision)
            .await
            .map_err(HttpRequestSendError::oauth)?;
        observed_revision = decision.revision;

        let token = provider
            .get_token()
            .await
            .map_err(HttpRequestSendError::oauth)?;
        let headers = bearer_headers(&token).map_err(|message| {
            HttpRequestSendError::oauth(OAuthClientError::ScopeEscalation(message))
        })?;
        retry_request = retry_request.headers(headers);
        request = retry_request;
    }
}

#[cfg(feature = "oauth-client")]
fn scope_challenge(headers: &reqwest::header::HeaderMap) -> Option<OAuthScopeChallenge> {
    headers
        .get_all(reqwest::header::WWW_AUTHENTICATE)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .find_map(OAuthScopeChallenge::from_www_authenticate)
}

/// Whether the response carries any Bearer challenge at all.
///
/// Wider than [`scope_challenge`], deliberately. That one accepts only
/// `insufficient_scope` with scopes attached, because a scope escalation needs
/// to know what to ask for. A `401` names no scopes and often no error code
/// either, and the remedy does not depend on which: the token was rejected, so
/// fetch another. Requiring a challenge at all is what separates "the server
/// is asking for authentication" from a bare `401` that some intermediary
/// produced, which is not ours to retry (#1370).
#[cfg(feature = "oauth-client")]
fn bearer_challenge(headers: &reqwest::header::HeaderMap) -> Option<OAuthBearerChallenge> {
    headers
        .get_all(reqwest::header::WWW_AUTHENTICATE)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .find_map(OAuthBearerChallenge::from_www_authenticate)
}

fn http_status_error(status: reqwest::StatusCode, headers: &reqwest::header::HeaderMap) -> String {
    #[cfg(feature = "oauth-client")]
    if scope_challenge(headers).is_some() {
        return format!("server returned HTTP {status}: insufficient_scope");
    }

    // A 401 that reached here survived the re-authentication retry, so the
    // replacement token was rejected too. Keep known RFC 6750 error codes,
    // but never echo descriptions, scope values, or metadata URLs into logs.
    // The original challenge remains in structured status data.
    #[cfg(feature = "oauth-client")]
    if let Some(challenge) = bearer_challenge(headers) {
        let detail = match challenge.error.as_deref() {
            Some("invalid_request") => "invalid_request",
            Some("invalid_token") => "invalid_token",
            Some("insufficient_scope") => "insufficient_scope",
            _ => "authentication rejected",
        };
        return format!("server returned HTTP {status}: {detail}");
    }

    #[cfg(not(feature = "oauth-client"))]
    let _ = headers;
    // A 301/302/303 reaches here because the redirect policy declines to
    // follow it. Only that is reported: the `Location` is a URL, which
    // diagnostics leave out.
    if matches!(
        status,
        reqwest::StatusCode::MOVED_PERMANENTLY
            | reqwest::StatusCode::FOUND
            | reqwest::StatusCode::SEE_OTHER
    ) {
        return format!(
            "server returned HTTP {status}: redirect not followed (only 307 and 308 keep \
             the request body; use the final URL)"
        );
    }
    format!("server returned HTTP {status}")
}

/// Why [`read_bounded_body`] failed to produce a body.
enum BoundedBodyError {
    /// The body exceeded the configured limit, caught either from a
    /// declared `Content-Length` or while streaming. The connection itself
    /// is fine: only this one response was oversized.
    TooLarge(String),
    /// The underlying HTTP stream failed while reading the body.
    Io(reqwest::Error),
}

/// Read a non-SSE response body, rejecting it once its size exceeds
/// `limit` instead of buffering an unbounded amount of memory.
///
/// Checks `Content-Length` up front so a body that declares itself too
/// large is rejected before a byte is read, then streams and counts bytes
/// so a chunked body (or one served with `Connection: close` and no
/// declared length) is bounded too. Decoded as UTF-8 lossily, the same way
/// the SSE path in this file decodes bytes (`String::from_utf8_lossy`).
async fn read_bounded_body(
    response: reqwest::Response,
    limit: usize,
) -> std::result::Result<String, BoundedBodyError> {
    if let Some(len) = response.content_length()
        && len > limit as u64
    {
        return Err(BoundedBodyError::TooLarge(format!(
            "response body of {len} bytes exceeds the {limit}-byte limit"
        )));
    }

    use futures::StreamExt;
    let mut stream = response.bytes_stream();
    let mut buf = Vec::new();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(BoundedBodyError::Io)?;
        buf.extend_from_slice(&chunk);
        if buf.len() > limit {
            return Err(BoundedBodyError::TooLarge(format!(
                "response body exceeds the {limit}-byte limit"
            )));
        }
    }
    Ok(String::from_utf8_lossy(&buf).into_owned())
}

/// Maximum bytes read for an HTTP error body before truncating it.
///
/// Error bodies are read to recognize MCP JSON-RPC error frames. An oversized
/// or unrecognized body falls back to status-only diagnostics; raw bodies are
/// never quoted in transport-generated errors or logs.
const MAX_ERROR_BODY_SIZE: usize = 64 * 1024;

/// Read a non-2xx response body up to [`MAX_ERROR_BODY_SIZE`], truncating
/// rather than failing the request if the server sends more.
async fn read_error_body(response: reqwest::Response) -> String {
    use futures::StreamExt;
    let mut stream = response.bytes_stream();
    let mut buf = Vec::new();
    while let Some(chunk) = stream.next().await {
        let Ok(chunk) = chunk else { break };
        buf.extend_from_slice(&chunk);
        if buf.len() > MAX_ERROR_BODY_SIZE {
            buf.truncate(MAX_ERROR_BODY_SIZE);
            break;
        }
    }
    String::from_utf8_lossy(&buf).into_owned()
}

fn operation_label(parsed: Option<&serde_json::Value>) -> String {
    let Some(method) = parsed
        .and_then(|value| value.get("method"))
        .and_then(serde_json::Value::as_str)
    else {
        return "unknown".to_string();
    };
    let target = match method {
        "tools/call" | "prompts/get" => parsed
            .and_then(|value| value.pointer("/params/name"))
            .and_then(serde_json::Value::as_str),
        "resources/read" => parsed
            .and_then(|value| value.pointer("/params/uri"))
            .and_then(serde_json::Value::as_str),
        "tasks/get" | "tasks/update" | "tasks/cancel" => parsed
            .and_then(|value| value.pointer("/params/taskId"))
            .and_then(serde_json::Value::as_str),
        _ => None,
    };
    match target {
        Some(target) => format!("{method}:{target}"),
        None => method.to_string(),
    }
}

#[async_trait]
impl ClientTransport for HttpClientTransport {
    async fn send(&mut self, message: &str) -> Result<()> {
        if !self.connected.load(Ordering::Acquire) {
            return Err(Error::Transport("Transport closed".to_string()));
        }

        // Notifications (frames without an `id`) are awaited inline (below) to
        // keep `notifications/initialized` ordered before the first request
        // (#967). That inline await blocks the whole message loop, so it must
        // be bounded independently: a server that stalls the notification's
        // 202 (observed: a multi-instance server holding the POST for the full
        // request timeout) would otherwise freeze the client with no output.
        let parsed_message = serde_json::from_str::<serde_json::Value>(message).ok();
        let is_notification = parsed_message
            .as_ref()
            .map(|v| v.get("id").is_none())
            .unwrap_or(false);
        let method = parsed_message
            .as_ref()
            .and_then(|value| value.get("method"))
            .and_then(serde_json::Value::as_str);
        // Only a request (a frame with both `method` and `id`) is owed a
        // response. The client's replies to server requests also pass through
        // here, and must not have error frames injected on their behalf.
        let reply_id = parsed_message
            .as_ref()
            .filter(|value| value.get("method").is_some())
            .and_then(|value| value.get("id"))
            .cloned();
        let operation = operation_label(parsed_message.as_ref());
        if method == Some("initialize") {
            self.initialize_request_id = reply_id
                .clone()
                .and_then(|value| serde_json::from_value(value).ok());
        }
        let outbound_version = parsed_message
            .as_ref()
            .and_then(|value| {
                value.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion")
            })
            .and_then(serde_json::Value::as_str)
            .map(str::to_string);
        let is_modern_request =
            outbound_version.as_deref() == Some(crate::protocol::PROTOCOL_VERSION_2026_07_28);
        if is_modern_request {
            self.protocol_version = outbound_version.clone();
            // Final requests are sessionless even when a transitional peer
            // incorrectly returns a legacy session header from discovery.
            self.session_id = None;
        }
        if is_modern_request
            && method == Some("tools/list")
            && let Some(id) = reply_id.as_ref()
        {
            let has_cursor = parsed_message
                .as_ref()
                .and_then(|value| value.pointer("/params/cursor"))
                .is_some_and(|cursor| !cursor.is_null());
            if self.pending_tools_lists.len() >= MAX_PENDING_TOOLS_LISTS {
                self.pending_tools_lists.remove(0);
            }
            self.pending_tools_lists.push((id.clone(), has_cursor));
        }
        let timeout = if is_notification {
            self.config
                .notification_timeout
                .min(self.config.request_timeout)
        } else {
            self.config.request_timeout
        };

        // Build request with headers
        let mut request = self
            .client
            .post(&self.url)
            .header("Content-Type", "application/json")
            .header("Accept", "application/json, text/event-stream");
        // A subscription is intentionally long-lived; its handle owns the
        // timeout/cancellation decision. Every ordinary request remains
        // bounded by the configured request timeout.
        if method != Some("subscriptions/listen") {
            request = request.timeout(timeout);
        }

        if !is_modern_request && let Some(ref session_id) = self.session_id {
            request = request.header("mcp-session-id", session_id);
        }

        if let Some(version) = outbound_version.as_ref().or(self.protocol_version.as_ref()) {
            request = request.header("mcp-protocol-version", version);
        }

        if let Some(method) = method {
            request = request.header(MCP_METHOD_HEADER, method);
            let name = match method {
                "tools/call" | "prompts/get" => parsed_message
                    .as_ref()
                    .and_then(|value| value.pointer("/params/name"))
                    .and_then(serde_json::Value::as_str),
                "resources/read" => parsed_message
                    .as_ref()
                    .and_then(|value| value.pointer("/params/uri"))
                    .and_then(serde_json::Value::as_str),
                "tasks/get" | "tasks/update" | "tasks/cancel" => parsed_message
                    .as_ref()
                    .and_then(|value| value.pointer("/params/taskId"))
                    .and_then(serde_json::Value::as_str),
                _ => None,
            };
            if let Some(name) = name {
                request = request.header(MCP_NAME_HEADER, name);
            }
        }

        if let Some(parsed) = parsed_message.as_ref() {
            for (name, value) in self.outgoing_custom_headers(parsed) {
                request = request.header(name, value);
            }
        }

        for (key, value) in &self.config.headers {
            request = request.header(key.as_str(), value.as_str());
        }

        #[cfg(feature = "oauth-client")]
        let initial_scope_revision = match &self.scope_escalation {
            Some(runtime) => runtime.state.lock().await.revision,
            None => 0,
        };

        // Dynamic token provider overrides static Authorization header
        #[cfg(feature = "oauth-client")]
        if let Some(ref provider) = self.token_provider {
            let token = provider
                .get_token()
                .await
                .map_err(|_| Error::Transport("Token provider failed".to_string()))?;
            request = request.headers(bearer_headers(&token).map_err(Error::Transport)?);
        }

        let request = request.body(message.to_string());

        // Notifications are awaited inline (bounded above) even after session
        // establishment, so `notifications/initialized` reaches the server
        // ahead of the first request rather than racing it on a pooled
        // connection, which strict servers rejected (#967).
        //
        // After session is established, send requests in a background task
        // so the message loop can continue processing incoming SSE messages.
        // This prevents a deadlock when the server blocks on a
        // bidirectional request (sampling/elicitation) that requires the
        // client to handle a request on the originating POST response stream.
        if !is_notification && (self.session_id.is_some() || is_modern_request) {
            let tx = self.incoming_tx.clone();
            // The caller is parked on this request id in the message loop's
            // correlation map. Every failure branch below delivers a frame
            // carrying it, so a background POST that dies (network error,
            // timeout, HTTP error, empty body, response stream closed early)
            // wakes the caller with an error instead of hanging it forever.
            //
            // A response (the client's reply to a server-initiated request)
            // also takes this path, to keep the message loop non-blocking
            // (#967), but `reply_id` is `None` for it: it is not tracked in
            // `request_tasks` below, and its failure branches log instead of
            // injecting an error frame, since the id belongs to the server's
            // request, not one of ours (#1466).
            let request_id = reply_id
                .clone()
                .and_then(|value| serde_json::from_value(value).ok());
            let is_subscription = method == Some("subscriptions/listen");
            // Whether this request carried `Mcp-Session-Id`: a 404 then means
            // the session expired (#1556). A 2026-07-28 request is sessionless,
            // and there a 404 is an ordinary status (an unknown method answers
            // 404 with -32601).
            let session_expiry_on_404 = !is_modern_request
                && self.session_id.is_some()
                && self.config.session_recovery
                && reply_id.is_some();
            let reply_id = reply_id.clone();
            let connected = self.connected.clone();
            let last_event_id = self.last_event_id.clone();
            let sse_retry_delay = self.sse_retry_delay.clone();
            let sse_reconnect_signal = self.sse_reconnect_signal.clone();
            let max_sse_event_size = self.config.max_sse_event_size;
            let max_response_body_size = self.config.max_response_body_size;
            let request_resource = self.url.clone();
            #[cfg(feature = "oauth-client")]
            let token_provider = self.token_provider.clone();
            #[cfg(feature = "oauth-client")]
            let scope_escalation = self.scope_escalation.clone();
            self.request_tasks.retain(|_, task| !task.is_finished());
            let task = tokio::spawn(async move {
                let response_result = send_http_request(
                    request,
                    &request_resource,
                    &operation,
                    #[cfg(feature = "oauth-client")]
                    token_provider,
                    #[cfg(feature = "oauth-client")]
                    scope_escalation,
                    #[cfg(feature = "oauth-client")]
                    initial_scope_revision,
                )
                .await;
                let response = match response_result {
                    Ok(r) => r,
                    Err(e) => {
                        let connection_failed = e.connection_failed;
                        tracing::error!(error = %e.message, "Background HTTP request failed");
                        if let Some(id) = &reply_id {
                            let _ = tx.send(transport_error_frame(id, &e.message)).await;
                        }
                        if connection_failed {
                            connected.store(false, Ordering::Release);
                        }
                        return;
                    }
                };

                let status = response.status();

                // 202 Accepted acknowledges a notification or a reply, never a
                // request, which is owed a JSON-RPC response.
                if status == reqwest::StatusCode::ACCEPTED {
                    if let Some(id) = &reply_id {
                        let _ = tx
                            .send(transport_error_frame(id, ACCEPTED_WITHOUT_RESPONSE))
                            .await;
                    }
                    return;
                }

                if !status.is_success() {
                    let status_error = http_status_error(status, response.headers());
                    let retry_after = retry_after_header(response.headers());
                    let www_authenticate = www_authenticate_header(response.headers());
                    let body = read_error_body(response).await;

                    // The spec's session-expired signal is the 404 itself, so
                    // it triggers recovery whatever the body holds: nothing, a
                    // -32005, or another code such as the -32001 other SDKs
                    // use. Hand the message loop the -32005 frame it already
                    // recovers on, and leave the transport open for the
                    // re-initialize.
                    if status == reqwest::StatusCode::NOT_FOUND
                        && session_expiry_on_404
                        && let Some(id) = &reply_id
                    {
                        tracing::info!("HTTP 404 on a session request, signalling session expiry");
                        let _ = tx.send(session_expired_frame(id)).await;
                        return;
                    }

                    // Forward a JSON-RPC error body so the message loop can
                    // detect -32005 (SessionNotFound) and trigger session
                    // recovery. If the server did not echo our request id (a
                    // null, absent, or foreign id such as a gateway's
                    // "server-error", on anything but the session-level
                    // -32005 signal), inject it so the awaiting caller is
                    // woken by this error instead of hanging. A response has
                    // no `reply_id`: its error body is forwarded only for the
                    // session signal, since any id it carries is the server's
                    // and could match one of the client's own pending
                    // requests.
                    if !body.is_empty()
                        && let Ok(mut v) = serde_json::from_str::<serde_json::Value>(&body)
                        && is_jsonrpc_error_response(&v)
                    {
                        let is_session_signal =
                            v.pointer("/error/code").and_then(|c| c.as_i64()) == Some(-32005);
                        if reply_id.is_none() && !is_session_signal {
                            tracing::warn!(status = %status, "reply POST returned a JSON-RPC error body");
                            return;
                        }
                        if !is_session_signal
                            && let Some(id) = &reply_id
                            && !v
                                .get("id")
                                .is_some_and(|actual| json_request_ids_match(actual, id))
                        {
                            v["id"] = id.clone();
                        }
                        if !is_session_signal && status_annotates_jsonrpc_error(status) {
                            merge_status_into_error(
                                &mut v,
                                status_error_data(status, &retry_after, &www_authenticate),
                            );
                        }
                        let _ = tx.send(v.to_string()).await;
                        return;
                    }

                    tracing::error!(status = %status, "HTTP error from server");
                    if let Some(id) = &reply_id {
                        let _ = tx
                            .send(transport_error_frame_with_data(
                                id,
                                &status_error,
                                status_error_data(status, &retry_after, &www_authenticate),
                            ))
                            .await;
                    }
                    // Only a status that says the endpoint is gone ends the
                    // transport. Anything else (a 429 from a rate limiter, a
                    // 502 or 503 from a gateway) failed this request alone
                    // and the next one may well succeed (#1553).
                    if status_ends_transport(status) {
                        connected.store(false, Ordering::Release);
                    }
                    return;
                }

                // Check if response is SSE-formatted
                let is_sse = response
                    .headers()
                    .get("content-type")
                    .and_then(|v| v.to_str().ok())
                    .is_some_and(|ct| ct.contains("text/event-stream"));

                if is_sse {
                    pump_post_sse(
                        response,
                        PostSseContext {
                            tx,
                            reply_id,
                            is_subscription,
                            is_modern_request,
                            connected,
                            last_event_id,
                            sse_retry_delay,
                            sse_reconnect_signal,
                            max_sse_event_size,
                        },
                    )
                    .await;
                } else {
                    // Non-SSE response: read body (bounded) and queue for recv().
                    match read_bounded_body(response, max_response_body_size).await {
                        Ok(body) if !body.is_empty() => {
                            let msgs = extract_json_messages(&body);
                            if msgs.is_empty() {
                                // A non-empty body that yields no JSON-RPC
                                // frames leaves the request uncorrelated. A
                                // response has no client request to wake;
                                // log instead.
                                if let Some(id) = &reply_id {
                                    let _ = tx
                                        .send(transport_error_frame(
                                            id,
                                            "server returned an unparseable response body",
                                        ))
                                        .await;
                                } else {
                                    tracing::warn!(
                                        "reply POST returned an unparseable response body"
                                    );
                                }
                            } else {
                                // A `{}` or non-JSON body still yields one
                                // "frame", which the message loop drops.
                                let answered = reply_id
                                    .as_ref()
                                    .is_none_or(|id| frames_answer_request(&msgs, id));
                                for msg in msgs {
                                    let _ = tx.send(msg).await;
                                }
                                if !answered && let Some(id) = &reply_id {
                                    let _ = tx
                                        .send(transport_error_frame(id, NO_RESPONSE_IN_BODY))
                                        .await;
                                }
                            }
                        }
                        Ok(_) => {
                            // 2xx with an empty body: no frame to correlate the
                            // request, so wake the caller rather than hang. A
                            // response has no client request to wake; log
                            // instead.
                            if let Some(id) = &reply_id {
                                let _ = tx
                                    .send(transport_error_frame(
                                        id,
                                        "server returned an empty response body",
                                    ))
                                    .await;
                            } else {
                                tracing::warn!("reply POST returned an empty response body");
                            }
                        }
                        Err(BoundedBodyError::TooLarge(message)) => {
                            // The connection itself is fine, only this
                            // reply was oversized: don't mark it
                            // disconnected. A response has no client
                            // request to wake; log instead.
                            tracing::error!(
                                error = %message,
                                "Background HTTP response body exceeded the size limit"
                            );
                            if let Some(id) = &reply_id {
                                let _ = tx.send(transport_error_frame(id, &message)).await;
                            }
                        }
                        Err(BoundedBodyError::Io(e)) => {
                            let message = request_error_message(&e);
                            tracing::error!(error = %message, "Failed to read response body");
                            if let Some(id) = &reply_id {
                                let _ = tx
                                    .send(transport_error_frame(
                                        id,
                                        &format!("failed to read response body: {message}"),
                                    ))
                                    .await;
                            }
                            connected.store(false, Ordering::Release);
                        }
                    }
                }
            });
            if let Some(request_id) = request_id {
                self.request_tasks.insert(request_id, task);
            }
            return Ok(());
        }

        // Pre-session (initialize) and notifications: handle synchronously.
        // For initialize this extracts session headers and starts the SSE
        // stream; for notifications the expected response is a bare 202.
        let response = send_http_request(
            request,
            &self.url,
            &operation,
            #[cfg(feature = "oauth-client")]
            self.token_provider.clone(),
            #[cfg(feature = "oauth-client")]
            self.scope_escalation.clone(),
            #[cfg(feature = "oauth-client")]
            initial_scope_revision,
        )
        .await
        .map_err(|e| Error::Transport(e.message))?;

        let status = response.status();

        // Extract session headers before consuming the body
        let new_session_id = response
            .headers()
            .get("mcp-session-id")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());
        let new_protocol_version = response
            .headers()
            .get("mcp-protocol-version")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        // 202 Accepted = notification acknowledged, no body
        if status == reqwest::StatusCode::ACCEPTED {
            // Still update session state if headers present
            if !is_modern_request && let Some(sid) = new_session_id {
                self.session_id = Some(sid);
            }
            if let Some(pv) = new_protocol_version {
                self.protocol_version = Some(pv);
            }
            // A request is owed a response; a 202 leaves its caller waiting.
            if let Some(id) = &reply_id {
                self.queue_incoming(transport_error_frame(id, ACCEPTED_WITHOUT_RESPONSE))
                    .await?;
            }
            return Ok(());
        }

        if !status.is_success() {
            let status_error = http_status_error(status, response.headers());
            let body = read_error_body(response).await;
            if is_modern_request
                && let Ok(mut error) = serde_json::from_str::<serde_json::Value>(&body)
                && is_jsonrpc_error_response(&error)
            {
                if let Some(id) = parsed_message.as_ref().and_then(|value| value.get("id"))
                    && !error
                        .get("id")
                        .is_some_and(|actual| json_request_ids_match(actual, id))
                {
                    error["id"] = id.clone();
                }
                self.queue_incoming(error.to_string()).await?;
                return Ok(());
            }
            // 404 only signals an expired session once a session exists.
            // Before that (the initial `initialize`), a 404 means the URL is
            // wrong, and reporting it as "Session expired" sends users
            // hunting session bugs instead of checking the endpoint path.
            if status == reqwest::StatusCode::NOT_FOUND
                && self.config.session_recovery
                && self.session_id.is_some()
            {
                return Err(Error::SessionExpired);
            }
            if status == reqwest::StatusCode::NOT_FOUND && self.session_id.is_none() {
                return Err(Error::Transport(
                    "HTTP 404: MCP endpoint not found (check the endpoint path; \
                     some servers serve MCP at the root, others at /mcp)"
                        .to_string(),
                ));
            }
            return Err(Error::Transport(status_error));
        }

        // Update session state
        if !is_modern_request && let Some(sid) = new_session_id {
            let is_new_session = self.session_id.is_none();
            self.session_id = Some(sid);

            if is_new_session && self.config.auto_sse {
                self.start_sse_stream();
            }
        }
        if let Some(pv) = new_protocol_version {
            self.protocol_version = Some(pv);
        }

        // A request answered with an SSE stream is read on its own task, as on
        // the session path, instead of to EOF here. The server may send a
        // request on that stream (`initialize` answered by a ping, say) and
        // hold the final reply until it is answered, and the message loop
        // that answers it is the caller of this method (#1558).
        let is_sse = response
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .is_some_and(|ct| ct.contains("text/event-stream"));
        if is_sse && reply_id.is_some() {
            let request_id = reply_id
                .clone()
                .and_then(|value| serde_json::from_value(value).ok());
            self.request_tasks.retain(|_, task| !task.is_finished());
            let task = tokio::spawn(pump_post_sse(
                response,
                PostSseContext {
                    tx: self.incoming_tx.clone(),
                    reply_id,
                    is_subscription: false,
                    is_modern_request,
                    connected: self.connected.clone(),
                    last_event_id: self.last_event_id.clone(),
                    sse_retry_delay: self.sse_retry_delay.clone(),
                    sse_reconnect_signal: self.sse_reconnect_signal.clone(),
                    max_sse_event_size: self.config.max_sse_event_size,
                },
            ));
            if let Some(request_id) = request_id {
                self.request_tasks.insert(request_id, task);
            }
            return Ok(());
        }

        // Read response body (bounded) and queue for recv()
        let body = match read_bounded_body(response, self.config.max_response_body_size).await {
            Ok(body) => body,
            Err(BoundedBodyError::TooLarge(message)) => {
                // The connection itself is fine, only this reply was
                // oversized: queue the error like #1465's fix does instead
                // of failing the transport, so the caller sees a transport
                // error rather than hanging.
                if let Some(id) = &reply_id {
                    self.queue_incoming(transport_error_frame(id, &message))
                        .await?;
                }
                return Ok(());
            }
            Err(BoundedBodyError::Io(e)) => {
                return Err(Error::Transport(format!(
                    "Failed to read response: {}",
                    request_error_message(&e)
                )));
            }
        };

        let msgs = extract_json_messages(&body);
        let answered = reply_id
            .as_ref()
            .is_none_or(|id| frames_answer_request(&msgs, id));
        let no_reply = if msgs.is_empty() {
            "server returned an empty response body"
        } else {
            NO_RESPONSE_IN_BODY
        };
        for msg in msgs {
            self.queue_incoming(msg).await?;
        }
        // Wake the caller rather than leave it parked on a reply that is not
        // coming. The transport itself is still healthy.
        if !answered && let Some(id) = &reply_id {
            self.queue_incoming(transport_error_frame(id, no_reply))
                .await?;
        }

        Ok(())
    }

    async fn recv(&mut self) -> Result<Option<String>> {
        match self.incoming_rx.recv().await {
            // All response paths converge here, including background final
            // POSTs. Normalize tools/list results on the transport-owning
            // task so validated x-mcp-header mappings are available to the
            // next tools/call without sharing mutable state across tasks.
            Some(msg) => Ok(Some(self.normalize_incoming_message(msg))),
            None => {
                self.connected.store(false, Ordering::Release);
                Ok(None)
            }
        }
    }

    fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Acquire)
    }

    async fn close(&mut self) -> Result<()> {
        self.connected.store(false, Ordering::Release);

        for (_, task) in self.request_tasks.drain() {
            task.abort();
        }

        // Abort SSE task
        if let Some(task) = self.sse_task.take() {
            task.abort();
        }

        // Send DELETE to terminate the session (best effort). Clearing
        // `session_id` up front, regardless of outcome, is what makes `Drop`
        // a no-op if it runs afterward.
        if let Some(session_id) = self.session_id.take() {
            send_session_delete(SessionDeleteParams {
                client: self.client.clone(),
                url: self.url.clone(),
                session_id,
                headers: self.config.headers.clone(),
                #[cfg(feature = "oauth-client")]
                token_provider: self.token_provider.clone(),
            })
            .await;
        }

        Ok(())
    }

    async fn reset_session(&mut self) {
        tracing::info!("Resetting session for re-initialization");

        for (_, task) in self.request_tasks.drain() {
            task.abort();
        }

        // Abort SSE task
        if let Some(task) = self.sse_task.take() {
            task.abort();
        }

        // Clear session state but keep the transport alive
        self.session_id = None;
        self.protocol_version = None;
        *self.last_event_id.write().await = None;
        *self.sse_retry_delay.write().await = None;

        // Drain any stale messages from the channel
        while self.incoming_rx.try_recv().is_ok() {}
    }

    async fn end_session(&mut self) {
        if let Some(session_id) = self.session_id.clone() {
            send_session_delete(SessionDeleteParams {
                client: self.client.clone(),
                url: self.url.clone(),
                session_id,
                headers: self.config.headers.clone(),
                #[cfg(feature = "oauth-client")]
                token_provider: self.token_provider.clone(),
            })
            .await;
        }
        self.reset_session().await;
    }

    fn supports_session_recovery(&self) -> bool {
        self.config.session_recovery
    }

    async fn cancel_request(&mut self, request_id: &RequestId) -> Result<()> {
        let mut in_flight = false;
        if let Some(task) = self.request_tasks.remove(request_id) {
            // Dropping reqwest's response byte stream closes this request's
            // HTTP response body, which is the final protocol's cancellation
            // signal. Other concurrent POST streams remain alive.
            in_flight = !task.is_finished();
            task.abort();
            let _ = task.await;
        }

        // On a 2025-11-25 session a closed connection is not a cancellation,
        // so the server keeps working unless it is told. The session id is
        // only held on that path; the final protocol is sessionless. The
        // spec forbids cancelling `initialize`.
        if in_flight
            && self.session_id.is_some()
            && self.initialize_request_id.as_ref() != Some(request_id)
        {
            let notification = serde_json::json!({
                "jsonrpc": "2.0",
                "method": "notifications/cancelled",
                "params": { "requestId": request_id },
            });
            // The POST is already aborted, so a failed notification leaves
            // nothing to retry; report success so the request is not kept
            // pending for a response that can no longer arrive.
            if let Err(error) = self.send(&notification.to_string()).await {
                tracing::warn!(%error, "Failed to send notifications/cancelled");
            }
        }
        Ok(())
    }
}

/// Best-effort teardown when a transport is dropped without an explicit
/// `close()`.
///
/// Aborts every in-flight task the transport owns. `close()` already drains
/// `request_tasks` and takes `sse_task`, so if `close()` ran first this is a
/// no-op; the fields are already empty. Also sends the session-termination
/// `DELETE`, spawned rather than awaited since `Drop::drop` cannot be async,
/// and only when a Tokio runtime is reachable from the current thread.
impl Drop for HttpClientTransport {
    fn drop(&mut self) {
        for (_, task) in self.request_tasks.drain() {
            task.abort();
        }
        if let Some(task) = self.sse_task.take() {
            task.abort();
        }

        // `session_id` is only `Some` for an established legacy (2025-11-25)
        // session; the 2026-07-28 stateless path never sets it (see
        // `is_modern_request` above), and `close()` clears it once its own
        // DELETE has been sent, which keeps this idempotent with `close()`.
        let Some(session_id) = self.session_id.take() else {
            return;
        };
        // Dropping outside a runtime (e.g. a transport built but never
        // connected) must not panic; there is nowhere to spawn the request.
        let Ok(handle) = tokio::runtime::Handle::try_current() else {
            return;
        };

        let params = SessionDeleteParams {
            client: self.client.clone(),
            url: self.url.clone(),
            session_id,
            headers: self.config.headers.clone(),
            #[cfg(feature = "oauth-client")]
            token_provider: self.token_provider.clone(),
        };
        handle.spawn(send_session_delete(params));
    }
}

// =============================================================================
// POST Response Streams
// =============================================================================

/// What [`pump_post_sse`] needs from the transport that spawned it.
struct PostSseContext {
    tx: mpsc::Sender<String>,
    /// The id of the request this stream answers, or `None` for a reply to a
    /// server request, which is owed nothing.
    reply_id: Option<serde_json::Value>,
    is_subscription: bool,
    is_modern_request: bool,
    connected: Arc<AtomicBool>,
    last_event_id: Arc<RwLock<Option<String>>>,
    sse_retry_delay: Arc<RwLock<Option<Duration>>>,
    sse_reconnect_signal: Arc<Notify>,
    max_sse_event_size: usize,
}

/// Forward the events of a POST's SSE response to the message loop as they
/// arrive, until the request's final reply or the end of the stream.
///
/// Every frame is queued as soon as its event is complete, so the client can
/// answer a server request sent on this stream (ping, sampling, elicitation)
/// before the final reply, which the server may be holding until it has.
/// A stream that ends without that reply wakes the caller with an error.
async fn pump_post_sse(response: reqwest::Response, ctx: PostSseContext) {
    let PostSseContext {
        tx,
        reply_id,
        is_subscription,
        is_modern_request,
        connected,
        last_event_id,
        sse_retry_delay,
        sse_reconnect_signal,
        max_sse_event_size,
    } = ctx;

    // Stream SSE response to extract id/retry fields
    let mut stream = response.bytes_stream();
    let mut parser = SseParser::with_limit(max_sse_event_size);
    let mut had_retry = false;
    let mut had_data = false;
    let mut subscription_acknowledged = false;

    use futures::StreamExt;
    while let Some(result) = stream.next().await {
        match result {
            Ok(bytes) => {
                let text = String::from_utf8_lossy(&bytes);
                let events = match parser.feed(&text) {
                    Ok(events) => events,
                    Err(e) => {
                        // A single event exceeded the cap;
                        // terminate the stream instead of
                        // buffering without bound. The
                        // response for this request is lost,
                        // so the transport is unusable.
                        tracing::error!(error = %e, "POST SSE stream terminated");
                        connected.store(false, Ordering::Release);
                        return;
                    }
                };
                for event in events {
                    if let Some(ref id) = event.id {
                        *last_event_id.write().await = Some(id.clone());
                    }
                    if let Some(retry_ms) = event.retry {
                        *sse_retry_delay.write().await = Some(Duration::from_millis(retry_ms));
                        had_retry = true;
                    }
                    if !event.data.is_empty() {
                        had_data = true;
                        let value = serde_json::from_str::<serde_json::Value>(&event.data);
                        let value = match value {
                            Ok(value) => value,
                            Err(error) if is_subscription => {
                                if let Some(id) = &reply_id {
                                    let _ = tx
                                        .send(transport_error_frame(
                                            id,
                                            &format!(
                                                "subscription stream returned invalid JSON: {error}"
                                            ),
                                        ))
                                        .await;
                                }
                                return;
                            }
                            Err(_) => {
                                let _ = tx.send(event.data).await;
                                continue;
                            }
                        };
                        let is_terminal = value.get("id").zip(reply_id.as_ref()).is_some_and(
                            |(actual, expected)| json_request_ids_match(actual, expected),
                        ) && (value.get("result").is_some()
                            || value.get("error").is_some());

                        if is_subscription {
                            let violation = if is_terminal {
                                if value.get("error").is_some() {
                                    None
                                } else if !subscription_acknowledged {
                                    Some("subscriptions/listen completed before acknowledgment")
                                } else if !value
                                    .pointer(
                                        "/result/_meta/io.modelcontextprotocol~1subscriptionId",
                                    )
                                    .zip(reply_id.as_ref())
                                    .is_some_and(|(actual, expected)| {
                                        json_request_ids_match(actual, expected)
                                    })
                                {
                                    Some(
                                        "subscriptions/listen result carried a missing or mismatched subscription ID",
                                    )
                                } else {
                                    None
                                }
                            } else if value.get("method").is_some() && value.get("id").is_none() {
                                let correlated = value
                                    .pointer(
                                        "/params/_meta/io.modelcontextprotocol~1subscriptionId",
                                    )
                                    .zip(reply_id.as_ref())
                                    .is_some_and(|(actual, expected)| {
                                        json_request_ids_match(actual, expected)
                                    });
                                let is_acknowledgment =
                                    value.get("method").and_then(serde_json::Value::as_str)
                                        == Some(notifications::SUBSCRIPTIONS_ACKNOWLEDGED);
                                if !correlated {
                                    Some(
                                        "subscription notification carried a missing or mismatched subscription ID",
                                    )
                                } else if !subscription_acknowledged && !is_acknowledgment {
                                    Some("subscription notification arrived before acknowledgment")
                                } else if subscription_acknowledged && is_acknowledgment {
                                    Some("subscription stream sent a duplicate acknowledgment")
                                } else {
                                    if is_acknowledgment {
                                        subscription_acknowledged = true;
                                    }
                                    None
                                }
                            } else {
                                Some("subscription stream returned an unrelated JSON-RPC message")
                            };
                            if let Some(message) = violation {
                                if let Some(id) = &reply_id {
                                    let _ = tx.send(transport_error_frame(id, message)).await;
                                }
                                return;
                            }
                        }
                        let _ = tx.send(event.data).await;
                        if is_terminal {
                            // The request is complete. Close the response
                            // body ourselves even if a non-conforming server
                            // leaves the SSE stream open after its final reply.
                            return;
                        }
                    }
                }
            }
            Err(e) => {
                tracing::warn!(error = %request_error_message(&e), "POST SSE stream error");
                break;
            }
        }
    }

    // If the POST SSE stream closed with a retry hint but no data,
    // the server expects us to reconnect the GET notification stream.
    // Signal the SSE loop to close its current stream and reconnect
    // with the updated last_event_id and sse_retry_delay.
    if !is_modern_request && had_retry && !had_data {
        sse_reconnect_signal.notify_one();
    } else {
        // The response stream closed without ever delivering a
        // terminal response. Acknowledgments and ordinary
        // notifications do not complete a request, so wake the
        // caller rather than leave it hanging. A response has
        // no client request to wake; log instead.
        let reason = if had_data {
            "server closed the response stream before the final reply"
        } else {
            "server closed the response stream without a reply"
        };
        if let Some(id) = &reply_id {
            let _ = tx.send(transport_error_frame(id, reason)).await;
        } else {
            tracing::warn!(reason, "reply POST stream closed unexpectedly");
        }
    }
}

// =============================================================================
// Session Termination
// =============================================================================

/// Owned inputs for the best-effort session-termination `DELETE`, shared by
/// `close()` (awaited inline) and `Drop` (spawned, since `drop()` cannot
/// await).
struct SessionDeleteParams {
    client: reqwest::Client,
    url: String,
    session_id: String,
    headers: HashMap<String, String>,
    #[cfg(feature = "oauth-client")]
    token_provider: Option<Arc<dyn TokenProvider>>,
}

/// Send the session-termination `DELETE`, best effort: errors and non-2xx
/// responses are ignored since the server will otherwise reclaim the session
/// on its own idle timeout.
async fn send_session_delete(params: SessionDeleteParams) {
    let SessionDeleteParams {
        client,
        url,
        session_id,
        headers,
        #[cfg(feature = "oauth-client")]
        token_provider,
    } = params;

    let mut request = client
        .delete(&url)
        .header("mcp-session-id", &session_id)
        .timeout(SESSION_DELETE_TIMEOUT);

    for (key, value) in &headers {
        request = request.header(key.as_str(), value.as_str());
    }

    // Dynamic token provider overrides the static Authorization header.
    #[cfg(feature = "oauth-client")]
    if let Some(ref provider) = token_provider
        && let Ok(token) = provider.get_token().await
        && let Ok(auth_headers) = bearer_headers(&token)
    {
        request = request.headers(auth_headers);
    }

    let _ = request.send().await;
}

// =============================================================================
// SSE Stream Background Loop
// =============================================================================

/// Parameters for the SSE background loop.
struct SseLoopParams {
    url: String,
    client: reqwest::Client,
    session_id: String,
    protocol_version: Option<String>,
    tx: mpsc::Sender<String>,
    last_event_id: Arc<RwLock<Option<String>>>,
    sse_retry_delay: Arc<RwLock<Option<Duration>>>,
    reconnect_signal: Arc<Notify>,
    connected: Arc<AtomicBool>,
    config: HttpClientConfig,
    #[cfg(feature = "oauth-client")]
    token_provider: Option<Arc<dyn TokenProvider>>,
}

/// Background loop that maintains the SSE stream connection.
///
/// Opens a GET request with `Accept: text/event-stream` and parses
/// incoming SSE events. Events are pushed into the mpsc channel for
/// `recv()` to return. Supports reconnection with `Last-Event-ID`.
async fn sse_stream_loop(params: SseLoopParams) {
    let SseLoopParams {
        url,
        client,
        session_id,
        protocol_version,
        tx,
        last_event_id,
        sse_retry_delay,
        reconnect_signal,
        connected,
        config,
        #[cfg(feature = "oauth-client")]
        token_provider,
    } = params;
    let mut reconnect_attempts = 0u32;

    loop {
        if !connected.load(Ordering::Acquire) {
            break;
        }

        let mut request = client
            .get(&url)
            .header("Accept", "text/event-stream")
            .header("mcp-session-id", &session_id);

        if let Some(ref version) = protocol_version {
            request = request.header("mcp-protocol-version", version);
        }

        for (key, value) in &config.headers {
            request = request.header(key.as_str(), value.as_str());
        }

        // Dynamic token provider overrides static Authorization header
        #[cfg(feature = "oauth-client")]
        if let Some(ref provider) = token_provider {
            match provider.get_token().await {
                Ok(token) => match bearer_headers(&token) {
                    Ok(headers) => request = request.headers(headers),
                    Err(error) => {
                        tracing::warn!(%error, "Token provider failed for SSE connection");
                        break;
                    }
                },
                Err(_) => {
                    tracing::warn!("Token provider failed for SSE connection");
                    break;
                }
            }
        }

        // Send Last-Event-ID for stream resumption
        if let Some(ref lei) = *last_event_id.read().await {
            request = request.header("Last-Event-ID", lei.clone());
        }

        let response = match request.send().await {
            Ok(r) if r.status().is_success() => {
                reconnect_attempts = 0;
                r
            }
            Ok(r) => {
                tracing::warn!(status = %r.status(), "SSE connection rejected");
                break;
            }
            Err(e) => {
                tracing::warn!(error = %request_error_message(&e), "SSE connection failed");
                if !config.sse_reconnect || reconnect_attempts >= config.max_sse_reconnect_attempts
                {
                    break;
                }
                reconnect_attempts += 1;
                let delay = sse_retry_delay
                    .read()
                    .await
                    .unwrap_or(config.sse_reconnect_delay);
                tokio::time::sleep(delay).await;
                continue;
            }
        };

        // Parse SSE stream, also listening for reconnect signals from POST handlers
        let mut stream = response.bytes_stream();
        let mut parser = SseParser::with_limit(config.max_sse_event_size);

        use futures::StreamExt;
        loop {
            tokio::select! {
                chunk = stream.next() => {
                    match chunk {
                        Some(Ok(bytes)) => {
                            let text = String::from_utf8_lossy(&bytes);
                            let events = match parser.feed(&text) {
                                Ok(events) => events,
                                Err(e) => {
                                    // A single event exceeded the cap;
                                    // terminate the stream (no reconnect,
                                    // the server would just repeat it)
                                    // instead of buffering without bound.
                                    tracing::error!(error = %e, "SSE stream terminated");
                                    connected.store(false, Ordering::Release);
                                    return;
                                }
                            };
                            for event in events {
                                if let Some(ref id) = event.id {
                                    *last_event_id.write().await = Some(id.clone());
                                }
                                if let Some(retry_ms) = event.retry {
                                    *sse_retry_delay.write().await = Some(Duration::from_millis(retry_ms));
                                }
                                if !event.data.is_empty() && tx.send(event.data).await.is_err() {
                                    return; // Channel closed, transport dropped
                                }
                            }
                        }
                        Some(Err(e)) => {
                            tracing::warn!(error = %request_error_message(&e), "SSE stream error");
                            break;
                        }
                        None => {
                            tracing::debug!("SSE stream ended");
                            break;
                        }
                    }
                }
                _ = reconnect_signal.notified() => {
                    tracing::debug!("SSE reconnect signal received, closing current stream");
                    break;
                }
            }
        }

        // Attempt reconnection
        if !config.sse_reconnect
            || !connected.load(Ordering::Acquire)
            || reconnect_attempts >= config.max_sse_reconnect_attempts
        {
            break;
        }
        reconnect_attempts += 1;
        let delay = sse_retry_delay
            .read()
            .await
            .unwrap_or(config.sse_reconnect_delay);
        tracing::info!(
            attempt = reconnect_attempts,
            max = config.max_sse_reconnect_attempts,
            delay_ms = delay.as_millis() as u64,
            "Reconnecting SSE stream"
        );
        tokio::time::sleep(delay).await;
    }
}

// =============================================================================
// SSE Parser
// =============================================================================

/// Extract JSON messages from a response body.
///
/// If the body is SSE-formatted (`event: message\ndata: ...\n\n`), extracts the
/// `data:` content from each event. Otherwise returns the body as-is.
/// Build a JSON-RPC error frame carrying `id`.
///
/// A post-session request POST is spawned in the background (so the message
/// loop can keep servicing the SSE stream), which means its failures happen
/// out of band from the caller. The caller is parked on the request id in the
/// message loop's correlation map; if the background POST fails before a real
/// response reaches the incoming channel, we must still deliver a frame with
/// this id or the caller hangs until the process exits. `-32000` is the
/// generic server-error code; the message names the transport-level cause.
fn transport_error_frame(id: &serde_json::Value, message: &str) -> String {
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": { "code": -32000, "message": message },
    })
    .to_string()
}

/// [`transport_error_frame`] with structured `data` attached to the error.
fn transport_error_frame_with_data(
    id: &serde_json::Value,
    message: &str,
    data: serde_json::Value,
) -> String {
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": { "code": -32000, "message": message, "data": data },
    })
    .to_string()
}

/// Whether a non-success status means the endpoint itself is gone, as opposed
/// to a transient failure of the one request it answered.
fn status_ends_transport(status: reqwest::StatusCode) -> bool {
    matches!(
        status,
        reqwest::StatusCode::NOT_FOUND | reqwest::StatusCode::GONE
    )
}

/// The raw `Retry-After` header value, if present and readable.
fn retry_after_header(headers: &reqwest::header::HeaderMap) -> Option<String> {
    headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|value| value.to_str().ok())
        .map(|value| value.trim().to_string())
}

/// The `WWW-Authenticate` challenge, if the response carries one. Several
/// header lines are joined with `, `, which RFC 9110 defines as equivalent.
fn www_authenticate_header(headers: &reqwest::header::HeaderMap) -> Option<String> {
    let values: Vec<&str> = headers
        .get_all(reqwest::header::WWW_AUTHENTICATE)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .map(str::trim)
        .collect();
    (!values.is_empty()).then(|| values.join(", "))
}

/// The `data` attached to the error for a non-success HTTP status; read it
/// back with [`HttpStatusDetails::from_error`].
fn status_error_data(
    status: reqwest::StatusCode,
    retry_after: &Option<String>,
    www_authenticate: &Option<String>,
) -> serde_json::Value {
    let mut data = serde_json::json!({ "httpStatus": status.as_u16() });
    if let Some(raw) = retry_after {
        data["retryAfter"] = serde_json::Value::String(raw.clone());
    }
    if let Some(challenge) = www_authenticate {
        data["wwwAuthenticate"] = serde_json::Value::String(challenge.clone());
    }
    data
}

/// Whether the status of a response with a JSON-RPC error body is recorded in
/// that error's `data`.
///
/// Limited to the statuses a caller acts on apart from the JSON-RPC code: an
/// authorization failure (401, 403) and a rate limit (429). Other statuses
/// that carry a JSON-RPC body (400 with -32602, 404 with -32601) are ordinary
/// in-band errors whose `data` stays exactly what the server sent.
fn status_annotates_jsonrpc_error(status: reqwest::StatusCode) -> bool {
    matches!(status.as_u16(), 401 | 403 | 429)
}

/// Record `status_data` (see [`status_error_data`]) in the `data` of the
/// JSON-RPC error in `frame`, keeping whatever the server put there.
///
/// - No `data`: it becomes `status_data`.
/// - An object: the status keys are added where absent, so a server that sent
///   its own `httpStatus` wins.
/// - Any other value (a string, an array): it moves under `serverData` next
///   to the status keys.
fn merge_status_into_error(frame: &mut serde_json::Value, status_data: serde_json::Value) {
    let Some(error) = frame
        .get_mut("error")
        .and_then(serde_json::Value::as_object_mut)
    else {
        return;
    };
    let data = error.entry("data").or_insert(serde_json::Value::Null);
    match data {
        serde_json::Value::Null => *data = status_data,
        serde_json::Value::Object(existing) => {
            if let serde_json::Value::Object(additions) = status_data {
                for (key, value) in additions {
                    existing.entry(key).or_insert(value);
                }
            }
        }
        other => {
            let original = other.take();
            let mut wrapped = status_data;
            wrapped["serverData"] = original;
            *other = wrapped;
        }
    }
}

/// The frame that tells the message loop the session behind request `id` is
/// gone: the `-32005` (SessionNotFound) error it recovers on.
fn session_expired_frame(id: &serde_json::Value) -> String {
    serde_json::json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": { "code": -32005, "message": "Session not found (HTTP 404)" },
    })
    .to_string()
}

/// The HTTP status of a non-success response, with its `Retry-After` and
/// `WWW-Authenticate` headers if the server sent them.
///
/// The failed request reaches the caller as an [`Error::JsonRpc`], and the
/// transport stays usable for the next request. The details ride in the
/// error's `data` as `{"httpStatus": 503, "retryAfter": "7"}`;
/// [`from_error`](Self::from_error) reads them back, for a retry policy that
/// needs to tell a transient failure from a dead server, or an authorization
/// failure from an ordinary error.
///
/// Two cases carry them:
///
/// - A status with no JSON-RPC error body (a 429 from a rate limiter, a 502
///   or 503 from a gateway). The error has code `-32000`.
/// - A 401, 403 or 429 whose body is a JSON-RPC error. That error is the
///   server's own, code and message included, with the status added to its
///   `data` (#1555): an existing object `data` keeps its keys and gains the
///   status keys where absent, and any other `data` moves under `serverData`.
///
/// A `404` or `410` with no JSON-RPC body still closes the transport, since
/// either means the endpoint is gone; a `404` on a session request instead
/// recovers the session. Errors from the initial `initialize` request, which
/// is sent synchronously, are reported as [`Error::Transport`] text and carry
/// no details.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HttpStatusDetails {
    /// The HTTP status code of the response.
    pub status: u16,
    /// The raw `Retry-After` header value: delta-seconds (`"7"`) or an
    /// HTTP-date, passed through unparsed.
    pub retry_after: Option<String>,
    /// The raw `WWW-Authenticate` challenge, as sent with a 401 or a 403.
    /// Several header lines are joined with `, `.
    pub www_authenticate: Option<String>,
}

impl HttpStatusDetails {
    /// Read the details from an error, if it came from a non-success HTTP
    /// status.
    pub fn from_error(error: &Error) -> Option<Self> {
        let Error::JsonRpc(error) = error else {
            return None;
        };
        let data = error.data.as_ref()?;
        let status = u16::try_from(data.get("httpStatus")?.as_u64()?).ok()?;
        let retry_after = data
            .get("retryAfter")
            .and_then(|value| value.as_str())
            .map(str::to_string);
        let www_authenticate = data
            .get("wwwAuthenticate")
            .and_then(|value| value.as_str())
            .map(str::to_string);
        Some(Self {
            status,
            retry_after,
            www_authenticate,
        })
    }

    /// The `Retry-After` value as a delay, when it is delta-seconds. `None`
    /// when absent or an HTTP-date.
    pub fn retry_after_delay(&self) -> Option<Duration> {
        self.retry_after
            .as_deref()?
            .parse::<u64>()
            .ok()
            .map(Duration::from_secs)
    }
}

/// Error for a request answered with `202 Accepted`, which acknowledges
/// notifications and replies but carries no response.
const ACCEPTED_WITHOUT_RESPONSE: &str =
    "server answered the request with 202 Accepted instead of a response";

/// Error for a response body that holds no reply to the request it answers.
const NO_RESPONSE_IN_BODY: &str = "server response did not contain a reply to this request";

/// Whether `frames` include the final response (a `result` or an `error`) to
/// the request with `id` (#1465).
fn frames_answer_request(frames: &[String], id: &serde_json::Value) -> bool {
    frames.iter().any(|frame| {
        serde_json::from_str::<serde_json::Value>(frame).is_ok_and(|value| {
            value
                .get("id")
                .is_some_and(|actual| json_request_ids_match(actual, id))
                && (value.get("result").is_some() || value.get("error").is_some())
        })
    })
}

fn json_request_ids_match(left: &serde_json::Value, right: &serde_json::Value) -> bool {
    left == right
        || (RequestId::from_json(left)
            .is_some_and(|left| Some(left) == RequestId::from_json(right)))
        || match (left, right) {
            (serde_json::Value::Number(number), serde_json::Value::String(value))
            | (serde_json::Value::String(value), serde_json::Value::Number(number)) => number
                .as_i64()
                .is_some_and(|number| value.parse::<i64>() == Ok(number)),
            _ => false,
        }
}

fn extract_json_messages(body: &str) -> Vec<String> {
    // A BOM ahead of the body would defeat both the SSE sniff below and the
    // JSON parse after it (#1303).
    let trimmed = crate::framing::clean_input_line(body);
    if trimmed.is_empty() {
        return Vec::new();
    }

    // Heuristic: SSE bodies start with "event:" or "data:" or "id:" or ":"
    let looks_like_sse = trimmed.starts_with("event:")
        || trimmed.starts_with("data:")
        || trimmed.starts_with("id:")
        || trimmed.starts_with(':');

    if looks_like_sse {
        // The body is already fully in memory here, so no event-size cap
        // applies: an unlimited parser never returns an error.
        let mut parser = SseParser::new();
        let events = parser.feed(body).unwrap_or_default();
        events.into_iter().map(|e| e.data).collect()
    } else {
        vec![trimmed.to_string()]
    }
}

/// A parsed SSE event.
#[derive(Debug)]
struct SseEvent {
    /// Event ID (from `id:` line), if present. String per SSE spec.
    id: Option<String>,
    /// Event data (from `data:` lines, joined with newlines).
    data: String,
    /// Server-requested retry delay in milliseconds (from `retry:` line).
    retry: Option<u64>,
}

/// Incremental SSE parser.
///
/// Handles partial chunks from the byte stream, buffering incomplete
/// lines across `feed()` calls. When constructed with
/// [`with_limit`](Self::with_limit), a single event whose buffered size
/// exceeds the limit terminates parsing with
/// [`Error::SseEventTooLarge`] instead of growing without bound
/// (rmcp #970 analog).
struct SseParser {
    /// Partial line buffer (when a chunk ends mid-line).
    buffer: String,
    /// Current event being parsed.
    current_id: Option<String>,
    current_data: Vec<String>,
    current_retry: Option<u64>,
    /// Total bytes across `current_data` lines, tracked incrementally.
    data_len: usize,
    /// Maximum buffered size for a single event.
    max_event_size: usize,
    /// Whether any chunk has been fed yet.
    ///
    /// A BOM is only a BOM at the very start of the stream; the same bytes
    /// later in it are a zero-width no-break space inside someone's data
    /// (#1303).
    at_stream_start: bool,
}

impl SseParser {
    /// Create a parser with no event-size limit.
    fn new() -> Self {
        Self::with_limit(usize::MAX)
    }

    /// Create a parser that rejects events buffering more than
    /// `max_event_size` bytes.
    fn with_limit(max_event_size: usize) -> Self {
        Self {
            buffer: String::new(),
            current_id: None,
            current_data: Vec::new(),
            current_retry: None,
            data_len: 0,
            max_event_size,
            at_stream_start: true,
        }
    }

    /// Feed a chunk of text and return any complete events.
    ///
    /// Returns [`Error::SseEventTooLarge`] when the bytes buffered for a
    /// single in-progress event exceed the configured limit. The parser
    /// should not be fed further after an error.
    fn feed(&mut self, text: &str) -> Result<Vec<SseEvent>> {
        // A leading BOM would make the first field name unrecognisable, so an
        // event whose first line is `data:` would be dropped whole (#1303).
        let text = if std::mem::take(&mut self.at_stream_start) {
            text.strip_prefix('\u{feff}').unwrap_or(text)
        } else {
            text
        };
        self.buffer.push_str(text);
        let mut events = Vec::new();

        // Process complete lines
        while let Some(newline_pos) = self.buffer.find('\n') {
            let line = self.buffer[..newline_pos]
                .trim_end_matches('\r')
                .to_string();
            self.buffer = self.buffer[newline_pos + 1..].to_string();

            if line.is_empty() {
                // Empty line = end of event
                if !self.current_data.is_empty() || self.current_retry.is_some() {
                    events.push(SseEvent {
                        id: self.current_id.take(),
                        data: self.current_data.join("\n"),
                        retry: self.current_retry.take(),
                    });
                    self.current_data.clear();
                    self.data_len = 0;
                }
                self.current_id = None;
                self.current_retry = None;
            } else if let Some(value) = line.strip_prefix("id:") {
                let trimmed = value.trim();
                if !trimmed.is_empty() {
                    self.current_id = Some(trimmed.to_string());
                }
            } else if let Some(value) = line.strip_prefix("data:") {
                let data = value.trim().to_string();
                self.data_len += data.len();
                self.current_data.push(data);
            } else if let Some(value) = line.strip_prefix("retry:") {
                self.current_retry = value.trim().parse().ok();
            }
            // Lines starting with ':' are comments (keep-alive) -- ignored
            // Lines starting with 'event:' are event types -- ignored (we only care about data)
        }

        // Everything still buffered belongs to a single unfinished event
        // (or an unfinished line of one). Cap it so a server that never
        // terminates an event can't grow the buffers without bound.
        let buffered = self.buffer.len() + self.data_len;
        if buffered > self.max_event_size {
            return Err(Error::SseEventTooLarge {
                size: buffered,
                limit: self.max_event_size,
            });
        }

        Ok(events)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    use crate::client::{
        OAuthAuthorizationAction, OAuthAuthorizationFlow, OAuthAuthorizationHandler,
        OAuthAuthorizationRequest, OAuthRedirectPolicy,
    };

    // =========================================================================
    // 401 re-authentication (#1370)
    // =========================================================================

    /// A provider that caches one token and only mints another after
    /// `invalidate`, which is what a real caching provider does and what makes
    /// re-sending a rejected token possible in the first place.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[derive(Clone, Default)]
    struct CachingProvider {
        minted: std::sync::Arc<std::sync::atomic::AtomicUsize>,
        invalidated: std::sync::Arc<std::sync::atomic::AtomicUsize>,
        cached: std::sync::Arc<tokio::sync::RwLock<Option<String>>>,
    }

    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[async_trait::async_trait]
    impl TokenProvider for CachingProvider {
        async fn get_token(&self) -> std::result::Result<String, OAuthClientError> {
            use std::sync::atomic::Ordering;
            if let Some(token) = self.cached.read().await.clone() {
                return Ok(token);
            }
            let mut cached = self.cached.write().await;
            if let Some(token) = cached.clone() {
                return Ok(token);
            }
            let token = format!("token-{}", self.minted.fetch_add(1, Ordering::SeqCst));
            *cached = Some(token.clone());
            Ok(token)
        }

        async fn invalidate(&self) {
            use std::sync::atomic::Ordering;
            self.invalidated.fetch_add(1, Ordering::SeqCst);
            *self.cached.write().await = None;
        }
    }

    /// Stands in for a provider that has not overridden `invalidate`: it can
    /// never replace its token, so the retry must stop rather than loop.
    #[cfg(feature = "oauth-client")]
    #[derive(Clone)]
    struct StuckProvider;

    #[cfg(feature = "oauth-client")]
    #[async_trait::async_trait]
    impl TokenProvider for StuckProvider {
        async fn get_token(&self) -> std::result::Result<String, OAuthClientError> {
            Ok("stale".to_string())
        }
    }

    /// A server that rejects `token-0` and `stale` with a `401` Bearer
    /// challenge and accepts anything else, so recovery shows up as a status
    /// change. `challenge` controls whether the `WWW-Authenticate` header is
    /// present at all.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    async fn unauthorizing_server(
        challenge: bool,
        seen: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    ) -> (String, tokio::task::JoinHandle<()>) {
        use axum::http::{HeaderMap, StatusCode};
        use axum::response::IntoResponse;
        use std::sync::atomic::Ordering;

        let app = axum::Router::new().route(
            "/",
            axum::routing::post(move |headers: HeaderMap| {
                let seen = seen.clone();
                async move {
                    seen.fetch_add(1, Ordering::SeqCst);
                    let accepted = headers
                        .get("authorization")
                        .and_then(|v| v.to_str().ok())
                        .is_some_and(|v| v != "Bearer token-0" && v != "Bearer stale");
                    if accepted {
                        return (StatusCode::OK, "{}").into_response();
                    }
                    if challenge {
                        (
                            StatusCode::UNAUTHORIZED,
                            [(
                                "www-authenticate",
                                r#"Bearer error="invalid_token", error_description="expired""#,
                            )],
                            "",
                        )
                            .into_response()
                    } else {
                        (StatusCode::UNAUTHORIZED, "").into_response()
                    }
                }
            }),
        );

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
        let handle = tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });
        (url, handle)
    }

    /// Drive `send_http_request` with an already-authorized request, the way
    /// the transport does.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    async fn send_with(
        url: &str,
        provider: std::sync::Arc<dyn TokenProvider>,
    ) -> reqwest::StatusCode {
        let token = provider.get_token().await.unwrap();
        let request = reqwest::Client::new()
            .post(url)
            .headers(bearer_headers(&token).unwrap());
        match send_http_request(request, url, "tools/call", Some(provider), None, 0).await {
            Ok(response) => response.status(),
            Err(_) => panic!("the request itself must not fail"),
        }
    }

    /// #1370: the reported case. An expired token is discarded and replaced
    /// rather than re-sent.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[tokio::test]
    async fn a_401_challenge_replaces_the_rejected_token_and_retries() {
        use std::sync::atomic::Ordering;
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let (url, _server) = unauthorizing_server(true, seen.clone()).await;
        let provider = CachingProvider::default();

        let status = send_with(&url, std::sync::Arc::new(provider.clone())).await;

        assert_eq!(status, reqwest::StatusCode::OK, "the retry must succeed");
        assert_eq!(
            provider.invalidated.load(Ordering::SeqCst),
            1,
            "without discarding it, the retry re-sends the rejected token"
        );
        assert_eq!(provider.minted.load(Ordering::SeqCst), 2);
        assert_eq!(seen.load(Ordering::SeqCst), 2, "exactly one retry");
    }

    /// The bound. A provider that cannot replace its token must not turn a
    /// `401` into a retry loop.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[tokio::test]
    async fn a_provider_that_cannot_re_authenticate_stops_after_one_retry() {
        use std::sync::atomic::Ordering;
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let (url, _server) = unauthorizing_server(true, seen.clone()).await;

        let status = send_with(&url, std::sync::Arc::new(StuckProvider)).await;

        assert_eq!(status, reqwest::StatusCode::UNAUTHORIZED);
        assert_eq!(
            seen.load(Ordering::SeqCst),
            2,
            "one original and one retry, and no more"
        );
    }

    /// A bare `401` with no challenge is somebody else's, commonly an
    /// intermediary. Retrying it would spend a token fetch on a server that
    /// never asked for one.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[tokio::test]
    async fn a_401_without_a_challenge_is_not_retried() {
        use std::sync::atomic::Ordering;
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let (url, _server) = unauthorizing_server(false, seen.clone()).await;
        let provider = CachingProvider::default();

        let status = send_with(&url, std::sync::Arc::new(provider.clone())).await;

        assert_eq!(status, reqwest::StatusCode::UNAUTHORIZED);
        assert_eq!(seen.load(Ordering::SeqCst), 1, "no retry");
        assert_eq!(provider.invalidated.load(Ordering::SeqCst), 0);
    }

    /// `invalidate` defaults to doing nothing, so a provider written before it
    /// existed still compiles and still behaves as it did.
    #[cfg(feature = "oauth-client")]
    #[tokio::test]
    async fn invalidate_defaults_to_a_no_op() {
        let provider = StuckProvider;
        provider.invalidate().await;
        assert_eq!(provider.get_token().await.unwrap(), "stale");
    }

    // =========================================================================
    // #1467: a real auth-code provider reauthorizes through invalidate()
    // =========================================================================

    /// Percent-decode an `application/x-www-form-urlencoded` body. `urlencoding`
    /// is already a required dependency of `oauth-client`, so this avoids
    /// pulling in `axum`'s `form` extractor (and `serde_urlencoded`) just for
    /// one test fixture.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    fn parse_form_body(body: &[u8]) -> HashMap<String, String> {
        String::from_utf8_lossy(body)
            .split('&')
            .filter(|pair| !pair.is_empty())
            .filter_map(|pair| {
                let mut parts = pair.splitn(2, '=');
                let key = urlencoding::decode(parts.next()?).ok()?.into_owned();
                let value = urlencoding::decode(parts.next().unwrap_or(""))
                    .ok()?
                    .into_owned();
                Some((key, value))
            })
            .collect()
    }

    /// One axum app standing in for both the protected resource and its
    /// authorization server, so [`OAuthAuthorizationFlow`] can discover,
    /// authorize, and refresh against real HTTP rather than a mock
    /// [`OAuthHttpClient`]. `/mcp` rejects everything except
    /// `refreshed-access-token`, so a successful retry proves the whole
    /// invalidate-then-refresh path rather than just that a token exists.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    async fn spawn_resource_server_fixture()
    -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
        use axum::body::Bytes;
        use axum::http::{HeaderMap, StatusCode};
        use axum::response::IntoResponse;
        use std::sync::atomic::Ordering;

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));

        let mcp_base = base.clone();
        let mcp_seen = seen.clone();
        let prm_base = base.clone();
        let as_base = base.clone();

        let app = axum::Router::new()
            .route(
                "/mcp",
                axum::routing::post(move |headers: HeaderMap| {
                    let seen = mcp_seen.clone();
                    let base = mcp_base.clone();
                    async move {
                        seen.fetch_add(1, Ordering::SeqCst);
                        let accepted = headers
                            .get("authorization")
                            .and_then(|v| v.to_str().ok())
                            .is_some_and(|v| v == "Bearer refreshed-access-token");
                        if accepted {
                            return (StatusCode::OK, "{}").into_response();
                        }
                        (
                            StatusCode::UNAUTHORIZED,
                            [(
                                "www-authenticate",
                                format!(
                                    r#"Bearer error="invalid_token", resource_metadata="{base}/.well-known/oauth-protected-resource""#
                                ),
                            )],
                            "",
                        )
                            .into_response()
                    }
                }),
            )
            .route(
                "/.well-known/oauth-protected-resource",
                axum::routing::get(move || {
                    let base = prm_base.clone();
                    async move {
                        axum::Json(serde_json::json!({
                            "resource": format!("{base}/mcp"),
                            "authorization_servers": [format!("{base}/as")],
                            "scopes_supported": ["mcp.scope"]
                        }))
                    }
                }),
            )
            .route(
                "/.well-known/oauth-authorization-server/as",
                axum::routing::get(move || {
                    let base = as_base.clone();
                    async move {
                        axum::Json(serde_json::json!({
                            "issuer": format!("{base}/as"),
                            "authorization_endpoint": format!("{base}/as/authorize"),
                            "token_endpoint": format!("{base}/as/token"),
                            "code_challenge_methods_supported": ["S256"],
                            "token_endpoint_auth_methods_supported": ["none"]
                        }))
                    }
                }),
            )
            .route(
                "/as/token",
                axum::routing::post(move |body: Bytes| async move {
                    let fields = parse_form_body(&body);
                    match fields.get("grant_type").map(String::as_str) {
                        Some("authorization_code") => axum::Json(serde_json::json!({
                            "access_token": "initial-access-token",
                            "token_type": "Bearer",
                            "expires_in": 3600,
                            "refresh_token": "refresh-token-1",
                            "scope": "mcp.scope"
                        }))
                        .into_response(),
                        Some("refresh_token") => axum::Json(serde_json::json!({
                            "access_token": "refreshed-access-token",
                            "token_type": "Bearer",
                            "expires_in": 3600,
                            "scope": "mcp.scope"
                        }))
                        .into_response(),
                        other => (
                            StatusCode::BAD_REQUEST,
                            format!("unexpected grant_type: {other:?}"),
                        )
                            .into_response(),
                    }
                }),
            );

        tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });
        (base, seen)
    }

    /// Returns [`OAuthAuthorizationAction::CallbackUrl`] immediately with a
    /// synthetic authorization code, so the flow completes without an actual
    /// browser redirect or loopback listener.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[derive(Clone, Default)]
    struct ImmediateCallbackHandler;

    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[async_trait::async_trait]
    impl OAuthAuthorizationHandler for ImmediateCallbackHandler {
        async fn authorize(
            &self,
            request: OAuthAuthorizationRequest,
        ) -> std::result::Result<OAuthAuthorizationAction, OAuthClientError> {
            let authorization_url = reqwest::Url::parse(&request.authorization_url).unwrap();
            let state = authorization_url
                .query_pairs()
                .find(|(name, _)| name == "state")
                .unwrap()
                .1
                .into_owned();
            let mut callback = reqwest::Url::parse(&request.redirect_uri).unwrap();
            callback
                .query_pairs_mut()
                .append_pair("code", "authorization-code")
                .append_pair("state", &state)
                .append_pair("iss", &request.issuer);
            Ok(OAuthAuthorizationAction::CallbackUrl(callback.to_string()))
        }
    }

    /// #1467 end-to-end: a real [`OAuthAuthorizationFlow`] behind the HTTP
    /// transport. The resource server 401s the token minted by the initial
    /// code exchange; the fix under test is that `invalidate()` now marks it
    /// unusable so `get_token()` performs an RFC 6749 refresh instead of
    /// handing back the same rejected token, and the retry succeeds.
    ///
    /// This exercises the real provider rather than a mock `TokenProvider`
    /// (unlike `a_401_challenge_replaces_the_rejected_token_and_retries`, which
    /// only proves the transport calls `invalidate()` before retrying, not that
    /// a real provider does anything useful when it does).
    ///
    /// `OAuthAuthorizationCode` is not used here instead: its only public entry
    /// point drives a real loopback callback listener (`start()` /
    /// `wait_for_callback()`), which has no existing test scaffolding to reuse
    /// across this module boundary and would not exercise anything beyond what
    /// this test and the provider-level refresh tests in `oauth_authcode.rs`
    /// already cover. `OAuthAuthorizationFlow` reaches the same real
    /// discovery/token-exchange/refresh code through its `CallbackUrl`
    /// authorization action, which completes without a real redirect.
    #[cfg(all(feature = "oauth-client", feature = "http"))]
    #[tokio::test]
    async fn a_real_auth_code_flow_provider_reauthorizes_through_invalidate() {
        use std::sync::atomic::Ordering;

        let (base, seen) = spawn_resource_server_fixture().await;
        let resource = format!("{base}/mcp");

        let flow = OAuthAuthorizationFlow::builder(&resource)
            .redirect_policy(OAuthRedirectPolicy::fixed(format!("{base}/callback")))
            .pre_registered_client("test-client", None)
            .authorization_handler(ImmediateCallbackHandler)
            .build()
            .unwrap();

        flow.authorize(["mcp.scope"]).await.unwrap();
        assert_eq!(
            flow.get_token().await.unwrap(),
            "initial-access-token",
            "sanity: the code exchange installed the token the fixture rejects"
        );

        let status = send_with(&resource, std::sync::Arc::new(flow.clone())).await;

        assert_eq!(status, reqwest::StatusCode::OK, "the retry must succeed");
        assert_eq!(
            flow.get_token().await.unwrap(),
            "refreshed-access-token",
            "invalidate() must have caused get_token() to refresh"
        );
        assert_eq!(
            seen.load(Ordering::SeqCst),
            3,
            "discovery probe, rejected first attempt, successful retry"
        );
    }

    // =========================================================================
    // Drop / session teardown (#1479)
    // =========================================================================

    /// Dropping a transport outside any Tokio runtime must not panic, even
    /// when it has a session id set, which is the one path that would
    /// otherwise try to spawn the teardown `DELETE`.
    #[test]
    fn drop_outside_a_runtime_does_not_panic() {
        let mut transport = HttpClientTransport::new("http://127.0.0.1:1");
        transport.session_id = Some("session-outside-runtime".to_string());
        transport.protocol_version = Some(crate::protocol::LATEST_PROTOCOL_VERSION.to_string());
        drop(transport);
    }

    // =========================================================================
    // SseParser tests
    // =========================================================================

    #[test]
    fn test_parse_complete_event() {
        let mut parser = SseParser::new();
        let events = parser
            .feed("id: 1\nevent: message\ndata: {\"hello\":\"world\"}\n\n")
            .unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].id, Some("1".to_string()));
        assert_eq!(events[0].data, "{\"hello\":\"world\"}");
    }

    /// #1303: a BOM ahead of the stream makes the first field name
    /// unrecognisable, so an event opening with `data:` is dropped whole.
    #[test]
    fn test_parse_strips_a_bom_at_the_start_of_the_stream() {
        let mut parser = SseParser::new();
        let events = parser
            .feed("\u{feff}data: {\"hello\":\"world\"}\n\n")
            .unwrap();

        assert_eq!(events.len(), 1, "the opening event must survive a BOM");
        assert_eq!(events[0].data, "{\"hello\":\"world\"}");
    }

    /// Only at the start. The same bytes later in the stream are a zero-width
    /// no-break space inside someone's data, and altering it would hand the
    /// caller content the server never sent.
    #[test]
    fn test_parse_keeps_a_bom_inside_the_stream() {
        let mut parser = SseParser::new();
        assert!(parser.feed("data: first\n\n").unwrap().len() == 1);

        let events = parser.feed("data: \u{feff}second\n\n").unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "\u{feff}second");
    }

    /// #1303: the same gap on the non-streaming path, where a BOM would
    /// defeat both the SSE sniff and the JSON parse behind it.
    #[test]
    fn test_extract_json_messages_strips_a_leading_bom() {
        let json = r#"{"jsonrpc":"2.0","id":1,"result":{}}"#;
        assert_eq!(
            extract_json_messages(&format!("\u{feff}{json}")),
            vec![json.to_string()],
        );

        let sse = extract_json_messages(&format!("\u{feff}data: {json}\n\n"));
        assert_eq!(
            sse,
            vec![json.to_string()],
            "a BOM must not hide an SSE body"
        );
    }

    #[test]
    fn test_parse_multiple_events() {
        let mut parser = SseParser::new();
        let events = parser
            .feed("id: 1\ndata: first\n\nid: 2\ndata: second\n\nid: 3\ndata: third\n\n")
            .unwrap();

        assert_eq!(events.len(), 3);
        assert_eq!(events[0].data, "first");
        assert_eq!(events[1].data, "second");
        assert_eq!(events[2].data, "third");
        assert_eq!(events[0].id, Some("1".to_string()));
        assert_eq!(events[1].id, Some("2".to_string()));
        assert_eq!(events[2].id, Some("3".to_string()));
    }

    #[test]
    fn test_parse_partial_chunks() {
        let mut parser = SseParser::new();

        // First chunk: partial event
        let events = parser.feed("id: 1\nda").unwrap();
        assert!(events.is_empty());

        // Second chunk: completes the event
        let events = parser.feed("ta: hello\n\n").unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].id, Some("1".to_string()));
        assert_eq!(events[0].data, "hello");
    }

    #[test]
    fn test_parse_multiline_data() {
        let mut parser = SseParser::new();
        let events = parser
            .feed("id: 1\ndata: line1\ndata: line2\ndata: line3\n\n")
            .unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "line1\nline2\nline3");
    }

    #[test]
    fn test_parse_comment_lines() {
        let mut parser = SseParser::new();
        let events = parser.feed(": keep-alive\nid: 1\ndata: hello\n\n").unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "hello");
    }

    #[test]
    fn test_parse_event_without_id() {
        let mut parser = SseParser::new();
        let events = parser.feed("data: no-id-event\n\n").unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].id, None);
        assert_eq!(events[0].data, "no-id-event");
    }

    #[test]
    fn test_empty_data_no_event() {
        let mut parser = SseParser::new();
        let events = parser.feed("id: 1\n\n").unwrap();

        // No data lines = no event produced
        assert!(events.is_empty());
    }

    #[test]
    fn test_parse_crlf_line_endings() {
        let mut parser = SseParser::new();
        let events = parser.feed("id: 1\r\ndata: crlf\r\n\r\n").unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].data, "crlf");
    }

    #[test]
    fn test_parse_json_data() {
        let mut parser = SseParser::new();
        let json = r#"{"jsonrpc":"2.0","method":"notifications/progress","params":{"token":"t1","progress":50}}"#;
        let input = format!("id: 42\nevent: message\ndata: {}\n\n", json);
        let events = parser.feed(&input).unwrap();

        assert_eq!(events.len(), 1);
        assert_eq!(events[0].id, Some("42".to_string()));

        // Verify it's valid JSON
        let parsed: serde_json::Value = serde_json::from_str(&events[0].data).unwrap();
        assert_eq!(parsed["method"], "notifications/progress");
    }

    #[test]
    fn test_event_exceeding_limit_is_rejected() {
        let mut parser = SseParser::with_limit(64);

        // An unterminated data line larger than the limit trips the cap.
        let big = "data: ".to_string() + &"x".repeat(128);
        let err = parser.feed(&big).unwrap_err();
        match err {
            Error::SseEventTooLarge { size, limit } => {
                assert!(size > 64, "size {} should exceed limit", size);
                assert_eq!(limit, 64);
            }
            other => panic!("expected SseEventTooLarge, got {:?}", other),
        }
    }

    #[test]
    fn test_accumulated_data_lines_count_toward_limit() {
        let mut parser = SseParser::with_limit(64);

        // Many complete data lines belonging to one unterminated event.
        let mut result = Ok(Vec::new());
        for _ in 0..10 {
            result = parser.feed("data: 0123456789\n");
            if result.is_err() {
                break;
            }
        }
        assert!(matches!(result, Err(Error::SseEventTooLarge { .. })));
    }

    #[test]
    fn test_events_within_limit_pass() {
        let mut parser = SseParser::with_limit(64);
        let events = parser.feed("data: hello\n\ndata: world\n\n").unwrap();
        assert_eq!(events.len(), 2);
    }

    // =========================================================================
    // Config tests
    // =========================================================================

    #[test]
    fn test_default_config() {
        let config = HttpClientConfig::default();
        assert!(config.auto_sse);
        assert_eq!(config.channel_capacity, 256);
        assert_eq!(config.request_timeout, Duration::from_secs(30));
        assert!(config.sse_reconnect);
        assert_eq!(config.sse_reconnect_delay, Duration::from_secs(1));
        assert_eq!(config.max_sse_reconnect_attempts, 5);
        assert!(config.headers.is_empty());
    }

    // =========================================================================
    // Transport constructor tests
    // =========================================================================

    #[test]
    fn test_new_transport() {
        let transport = HttpClientTransport::new("http://localhost:3000");
        assert_eq!(transport.url, "http://localhost:3000");
        assert!(transport.session_id.is_none());
        assert!(transport.protocol_version.is_none());
        assert!(transport.is_connected());
    }

    #[test]
    fn test_with_config() {
        let config = HttpClientConfig {
            request_timeout: Duration::from_secs(60),
            sse_reconnect: false,
            ..Default::default()
        };
        let transport = HttpClientTransport::with_config("http://example.com", config);
        assert_eq!(transport.url, "http://example.com");
        assert_eq!(transport.config.request_timeout, Duration::from_secs(60));
        assert!(!transport.config.sse_reconnect);
    }

    #[test]
    fn test_with_client() {
        let client = reqwest::Client::new();
        let transport = HttpClientTransport::with_client("http://example.com", client);
        assert_eq!(transport.url, "http://example.com");
        assert!(transport.is_connected());
    }

    // =========================================================================
    // Auth builder tests
    // =========================================================================

    #[test]
    fn test_bearer_token() {
        let transport =
            HttpClientTransport::new("http://localhost:3000").bearer_token("sk-test-token");
        assert_eq!(
            transport.config.headers.get("Authorization").unwrap(),
            "Bearer sk-test-token"
        );
    }

    #[test]
    fn test_api_key() {
        let transport = HttpClientTransport::new("http://localhost:3000").api_key("sk-api-key-123");
        assert_eq!(
            transport.config.headers.get("Authorization").unwrap(),
            "Bearer sk-api-key-123"
        );
    }

    #[test]
    fn test_api_key_header() {
        let transport =
            HttpClientTransport::new("http://localhost:3000").api_key_header("X-API-Key", "my-key");
        assert_eq!(transport.config.headers.get("X-API-Key").unwrap(), "my-key");
        assert!(!transport.config.headers.contains_key("Authorization"));
    }

    #[test]
    fn test_basic_auth() {
        let transport =
            HttpClientTransport::new("http://localhost:3000").basic_auth("admin", "secret");
        let header = transport.config.headers.get("Authorization").unwrap();
        assert!(header.starts_with("Basic "));
        use base64::Engine;
        let decoded = base64::engine::general_purpose::STANDARD
            .decode(header.strip_prefix("Basic ").unwrap())
            .unwrap();
        assert_eq!(String::from_utf8(decoded).unwrap(), "admin:secret");
    }

    #[test]
    fn test_custom_header() {
        let transport = HttpClientTransport::new("http://localhost:3000")
            .header("X-Custom", "value1")
            .header("X-Another", "value2");
        assert_eq!(transport.config.headers.get("X-Custom").unwrap(), "value1");
        assert_eq!(transport.config.headers.get("X-Another").unwrap(), "value2");
    }

    #[test]
    fn test_chaining_with_config() {
        let config = HttpClientConfig {
            request_timeout: Duration::from_secs(60),
            ..Default::default()
        };
        let transport =
            HttpClientTransport::with_config("http://localhost:3000", config).bearer_token("tk");
        assert_eq!(transport.config.request_timeout, Duration::from_secs(60));
        assert_eq!(
            transport.config.headers.get("Authorization").unwrap(),
            "Bearer tk"
        );
    }

    #[test]
    fn test_last_auth_wins() {
        let transport = HttpClientTransport::new("http://localhost:3000")
            .bearer_token("token1")
            .basic_auth("user", "pass");
        let header = transport.config.headers.get("Authorization").unwrap();
        assert!(header.starts_with("Basic "));
    }

    #[test]
    fn test_config_bearer_token() {
        let config = HttpClientConfig::default().bearer_token("tk-123");
        assert_eq!(
            config.headers.get("Authorization").unwrap(),
            "Bearer tk-123"
        );
    }

    #[test]
    fn test_config_header() {
        let config = HttpClientConfig::default().header("X-Foo", "bar");
        assert_eq!(config.headers.get("X-Foo").unwrap(), "bar");
    }

    #[test]
    fn test_config_api_key_header() {
        let config = HttpClientConfig::default().api_key_header("X-Key", "secret");
        assert_eq!(config.headers.get("X-Key").unwrap(), "secret");
    }

    #[test]
    fn test_config_basic_auth() {
        let config = HttpClientConfig::default().basic_auth("user", "pw");
        let header = config.headers.get("Authorization").unwrap();
        assert!(header.starts_with("Basic "));
    }

    #[test]
    fn sep_2243_encodes_only_unsafe_values() {
        assert_eq!(encode_header_value("us west 1"), "us west 1");
        assert_eq!(encode_header_value(""), "");
        assert_eq!(encode_header_value(" padded "), "=?base64?IHBhZGRlZCA=?=");
        assert_eq!(
            encode_header_value("Hello, 世界"),
            "=?base64?SGVsbG8sIOS4lueVjA==?="
        );
    }

    #[test]
    fn oauth_error_body_is_not_misclassified_as_jsonrpc() {
        assert!(!is_jsonrpc_error_response(&serde_json::json!({
            "error": "insufficient_scope",
            "error_description": "Token has insufficient scope"
        })));
        assert!(is_jsonrpc_error_response(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "error": {
                "code": -32022,
                "message": "Unsupported protocol version"
            }
        })));
    }

    #[test]
    fn sep_2243_validates_custom_header_annotations() {
        let mappings = custom_header_mappings(&serde_json::json!({
            "type": "object",
            "properties": {
                "region": {"type": "string", "x-mcp-header": "Region"},
                "priority": {"type": "integer", "x-mcp-header": "Priority"},
                "enabled": {"type": ["boolean", "null"], "x-mcp-header": "Enabled"}
            }
        }))
        .unwrap();
        assert_eq!(mappings.len(), 3);

        for invalid in [
            // `number` is not an allowed header type (SEP-2243), alone or
            // combined with null.
            serde_json::json!({
                "type": "object",
                "properties": {"ratio": {"type": "number", "x-mcp-header": "Ratio"}}
            }),
            serde_json::json!({
                "type": "object",
                "properties": {"ratio": {"type": ["number", "null"], "x-mcp-header": "Ratio"}}
            }),
            serde_json::json!({
                "type": "object",
                "properties": {"value": {"type": "object", "x-mcp-header": "Value"}}
            }),
            serde_json::json!({
                "type": "object",
                "properties": {
                    "a": {"type": "string", "x-mcp-header": "Region"},
                    "b": {"type": "string", "x-mcp-header": "region"}
                }
            }),
            serde_json::json!({
                "type": "object",
                "properties": {"value": {"type": "string", "x-mcp-header": "Bad Header"}}
            }),
        ] {
            assert!(custom_header_mappings(&invalid).is_err());
        }
    }

    #[test]
    fn sep_2243_filters_invalid_tools_and_caches_valid_mappings() {
        let mut transport = HttpClientTransport::new("http://localhost:3000");
        transport.protocol_version = Some(crate::protocol::PROTOCOL_VERSION_2026_07_28.to_string());
        let normalized = transport.normalize_incoming_message(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": 1,
                "result": {
                    "tools": [
                        {
                            "name": "valid",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "region": {"type": "string", "x-mcp-header": "Region"}
                                }
                            }
                        },
                        {
                            "name": "invalid",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "value": {"type": "array", "x-mcp-header": "Value"}
                                }
                            }
                        },
                        {
                            "name": "number_typed",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "ratio": {"type": "number", "x-mcp-header": "Ratio"}
                                }
                            }
                        }
                    ]
                }
            })
            .to_string(),
        );
        let parsed: serde_json::Value = serde_json::from_str(&normalized).unwrap();
        assert_eq!(parsed["result"]["tools"].as_array().unwrap().len(), 1);
        assert!(transport.tool_header_mappings.contains_key("valid"));
        assert!(!transport.tool_header_mappings.contains_key("invalid"));
        assert!(!transport.tool_header_mappings.contains_key("number_typed"));
    }

    fn tools_page(id: i64, tool: &str) -> String {
        serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": {
                "tools": [{
                    "name": tool,
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "region": {"type": "string", "x-mcp-header": "Region"}
                        }
                    }
                }]
            }
        })
        .to_string()
    }

    /// Only a first-page `tools/list` response replaces the mappings; a page
    /// requested with a cursor adds to them, and `tools/list_changed` drops
    /// them (#1589).
    #[test]
    fn tools_list_pages_merge_header_mappings_until_list_changed() {
        let mut transport = HttpClientTransport::new("http://localhost:3000");
        transport.protocol_version = Some(crate::protocol::PROTOCOL_VERSION_2026_07_28.to_string());

        transport
            .pending_tools_lists
            .push((serde_json::json!(1), false));
        transport.normalize_incoming_message(tools_page(1, "first"));
        transport
            .pending_tools_lists
            .push((serde_json::json!(2), true));
        transport.normalize_incoming_message(tools_page(2, "second"));
        assert!(transport.pending_tools_lists.is_empty());
        assert!(transport.tool_header_mappings.contains_key("first"));
        assert!(transport.tool_header_mappings.contains_key("second"));

        // A new first page replaces what was learned before.
        transport
            .pending_tools_lists
            .push((serde_json::json!(3), false));
        transport.normalize_incoming_message(tools_page(3, "third"));
        assert_eq!(transport.tool_header_mappings.len(), 1);
        assert!(transport.tool_header_mappings.contains_key("third"));

        transport
            .pending_tools_lists
            .push((serde_json::json!(4), true));
        transport.normalize_incoming_message(tools_page(4, "fourth"));
        assert_eq!(transport.tool_header_mappings.len(), 2);

        let notification = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/tools/list_changed"
        })
        .to_string();
        assert_eq!(
            transport.normalize_incoming_message(notification.clone()),
            notification
        );
        assert!(transport.tool_header_mappings.is_empty());
    }

    /// An error response for a pending `tools/list` still consumes its entry.
    #[test]
    fn tools_list_error_response_clears_pending_entry() {
        let mut transport = HttpClientTransport::new("http://localhost:3000");
        transport.protocol_version = Some(crate::protocol::PROTOCOL_VERSION_2026_07_28.to_string());
        transport
            .pending_tools_lists
            .push((serde_json::json!(7), true));
        transport.normalize_incoming_message(
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": 7,
                "error": {"code": -32603, "message": "boom"}
            })
            .to_string(),
        );
        assert!(transport.pending_tools_lists.is_empty());
    }

    // =========================================================================
    // Response vs. request classification (#1466)
    // =========================================================================

    /// A response (id present, no method) -- the client's reply to a
    /// server-initiated request -- must never be tracked in `request_tasks`,
    /// even when its id collides with one of the client's own request ids.
    /// A request with the same id is tracked as before, and `cancel_request`
    /// on that id only ever touches the client's own request.
    #[tokio::test]
    async fn responses_are_not_tracked_in_request_tasks() {
        let mut transport = HttpClientTransport::new("http://127.0.0.1:1");
        transport.session_id = Some("test-session".to_string());

        // The client's reply to a server request sharing id 7 with one of
        // the client's own (hypothetical) requests below.
        transport
            .send(r#"{"jsonrpc":"2.0","id":7,"result":{}}"#)
            .await
            .unwrap();
        assert!(
            transport.request_tasks.is_empty(),
            "a response must never be tracked in request_tasks"
        );

        // The client's own request using the same id is tracked normally.
        transport
            .send(r#"{"jsonrpc":"2.0","id":7,"method":"ping","params":{}}"#)
            .await
            .unwrap();
        assert!(
            transport.request_tasks.contains_key(&RequestId::Number(7)),
            "a request must still be tracked in request_tasks"
        );

        // cancel_request on the colliding id only ever aborts the client's
        // own request POST, never a reply (which was never inserted).
        transport
            .cancel_request(&RequestId::Number(7))
            .await
            .unwrap();
        assert!(transport.request_tasks.is_empty());
    }

    // =========================================================================
    // Redirect policy (#1475)
    // =========================================================================

    fn url(s: &str) -> reqwest::Url {
        reqwest::Url::parse(s).unwrap()
    }

    #[test]
    fn only_body_preserving_redirect_statuses_are_followed() {
        use reqwest::StatusCode;
        assert!(redirect_preserves_request(StatusCode::TEMPORARY_REDIRECT));
        assert!(redirect_preserves_request(StatusCode::PERMANENT_REDIRECT));
        for status in [
            StatusCode::MOVED_PERMANENTLY,
            StatusCode::FOUND,
            StatusCode::SEE_OTHER,
        ] {
            assert!(!redirect_preserves_request(status), "{status}");
        }
    }

    #[test]
    fn same_origin_redirect_is_followed_by_default() {
        assert!(should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("https://example.com/other"),
            false,
        ));
    }

    #[test]
    fn different_port_is_not_same_origin_by_default() {
        assert!(!should_follow_redirect(
            &url("https://example.com:8443/mcp"),
            &url("https://example.com:9443/mcp"),
            false,
        ));
    }

    #[test]
    fn default_port_matches_an_explicit_default_port() {
        // `https` defaults to 443, so an explicit `:443` is the same origin
        // as one with the port omitted.
        assert!(should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("https://example.com:443/other"),
            false,
        ));
    }

    #[test]
    fn cross_host_redirect_is_not_followed_by_default() {
        assert!(!should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("https://attacker.example/mcp"),
            false,
        ));
    }

    #[test]
    fn cross_host_redirect_is_followed_when_opted_in() {
        assert!(should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("https://attacker.example/mcp"),
            true,
        ));
    }

    #[test]
    fn https_to_http_downgrade_is_never_followed_even_when_opted_in() {
        assert!(!should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("http://example.com/mcp"),
            false,
        ));
        assert!(!should_follow_redirect(
            &url("https://example.com/mcp"),
            &url("http://attacker.example/mcp"),
            true,
        ));
    }

    #[test]
    fn http_to_https_redirect_is_not_a_downgrade() {
        // Not a downgrade, but still a scheme change, so it needs the same
        // opt-in as any other cross-origin redirect.
        assert!(!should_follow_redirect(
            &url("http://example.com/mcp"),
            &url("https://example.com/mcp"),
            false,
        ));
        assert!(should_follow_redirect(
            &url("http://example.com/mcp"),
            &url("https://example.com/mcp"),
            true,
        ));
    }
}