turul-mcp-client 0.4.2

//! HTTP transport implementation for MCP client

use async_trait::async_trait;
use futures::Stream;
use reqwest::{Client, Response};
use serde_json::{Deserializer, Value};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};
use url::Url;

use crate::error::{McpClientResult, TransportError};
use crate::transport::{
    ConnectionInfo, EventReceiver, ServerEvent, Transport, TransportCapabilities,
    TransportStatistics, TransportType,
};

/// Accept header for MCP POST requests per spec (MUST include both media types)
const MCP_POST_ACCEPT: &str = "application/json, text/event-stream";

/// HTTP transport for MCP client (Streamable HTTP)
#[derive(Debug)]
pub struct HttpTransport {
    /// HTTP client
    client: Client,
    /// Server endpoint URL
    endpoint: Url,
    /// Connection state
    connected: AtomicBool,
    /// Request counter
    request_counter: AtomicU64,
    /// Statistics
    stats: Arc<parking_lot::Mutex<TransportStatistics>>,
    /// Event sender for server events (Mutex so hot-path methods can take &self)
    event_sender: parking_lot::Mutex<Option<mpsc::UnboundedSender<ServerEvent>>>,
    /// Queue for progress notifications when no listener exists
    queued_events: Arc<parking_lot::Mutex<Vec<ServerEvent>>>,
    /// Session ID from server (set after initialization) - shared between transport and SSE task
    session_id: Arc<parking_lot::Mutex<Option<String>>>,
    /// Per-request `Authorization` header override.
    ///
    /// Reqwest's `default_headers` are baked into the `Client` at construction
    /// (see `with_config`); they cannot be mutated thereafter without rebuilding
    /// the `Client` (and dropping the connection pool). To support bearer
    /// rotation on a long-lived transport — see [`Transport::update_auth_header`] —
    /// we store the live override here and inject it on every outbound request
    /// via `RequestBuilder::header(…)`, which overrides any same-named entry in
    /// `default_headers`. Shared with the SSE GET listener task.
    auth_override: Arc<parking_lot::RwLock<Option<String>>>,
    /// Negotiated MCP spec version, sent as the `MCP-Protocol-Version` header.
    /// Defaults to 2025-11-25; the client sets it after negotiation so a 2026-07-28
    /// connection advertises 2026-07-28 (the schema requires the header to match the
    /// per-request `_meta` protocolVersion). A 2026-07-28 (stateless) connection also
    /// stops sending the removed `Mcp-Session-Id` header.
    protocol_version: Arc<parking_lot::RwLock<String>>,
}

impl HttpTransport {
    /// Create a new HTTP transport
    pub fn new(endpoint: &str) -> McpClientResult<Self> {
        let url = Url::parse(endpoint)
            .map_err(|e| TransportError::ConnectionFailed(format!("Invalid URL: {}", e)))?;

        // Validate URL scheme
        if !matches!(url.scheme(), "http" | "https") {
            return Err(TransportError::ConnectionFailed(format!(
                "Invalid scheme for HTTP transport: {}",
                url.scheme()
            ))
            .into());
        }

        let client = Client::builder()
            .timeout(Duration::from_secs(30))
            .user_agent("mcp-client/0.1.0")
            // HTTP/2 keepalives: proactively PING so silent server/intermediary drops
            // surface immediately instead of on the next real request. No-op on h1.
            .http2_keep_alive_interval(Duration::from_secs(30))
            .http2_keep_alive_timeout(Duration::from_secs(10))
            .http2_keep_alive_while_idle(true)
            .build()
            .map_err(|e| TransportError::Http(format!("Failed to create HTTP client: {}", e)))?;

        Ok(Self {
            client,
            endpoint: url,
            connected: AtomicBool::new(false),
            request_counter: AtomicU64::new(0),
            stats: Arc::new(parking_lot::Mutex::new(TransportStatistics::default())),
            event_sender: parking_lot::Mutex::new(None),
            queued_events: Arc::new(parking_lot::Mutex::new(Vec::new())),
            session_id: Arc::new(parking_lot::Mutex::new(None)),
            auth_override: Arc::new(parking_lot::RwLock::new(None)),
            protocol_version: Arc::new(parking_lot::RwLock::new("2025-11-25".to_string())),
        })
    }

    /// Create HTTP transport with connection configuration applied
    pub fn with_config(
        endpoint: &str,
        config: &crate::config::ConnectionConfig,
    ) -> McpClientResult<Self> {
        let url = Url::parse(endpoint)
            .map_err(|e| TransportError::ConnectionFailed(format!("Invalid URL: {}", e)))?;

        if !matches!(url.scheme(), "http" | "https") {
            return Err(TransportError::ConnectionFailed(format!(
                "Invalid scheme for HTTP transport: {}",
                url.scheme()
            ))
            .into());
        }

        let user_agent = config.user_agent.as_deref().unwrap_or("mcp-client/0.1.0");

        let mut builder = Client::builder()
            .timeout(Duration::from_secs(30))
            .user_agent(user_agent)
            .pool_max_idle_per_host(config.pool_settings.max_idle_per_host as usize)
            .pool_idle_timeout(config.pool_settings.idle_timeout)
            // HTTP/2 keepalives: proactively PING so silent server/intermediary drops
            // surface immediately instead of on the next real request. No-op on h1.
            .http2_keep_alive_interval(Duration::from_secs(30))
            .http2_keep_alive_timeout(Duration::from_secs(10))
            .http2_keep_alive_while_idle(true);

        builder = if config.follow_redirects {
            builder.redirect(reqwest::redirect::Policy::limited(
                config.max_redirects as usize,
            ))
        } else {
            builder.redirect(reqwest::redirect::Policy::none())
        };

        if let Some(ref headers) = config.headers {
            let mut header_map = reqwest::header::HeaderMap::new();
            for (k, v) in headers {
                if let (Ok(name), Ok(value)) = (
                    reqwest::header::HeaderName::from_bytes(k.as_bytes()),
                    reqwest::header::HeaderValue::from_str(v),
                ) {
                    header_map.insert(name, value);
                }
            }
            builder = builder.default_headers(header_map);
        }

        let client = builder
            .build()
            .map_err(|e| TransportError::Http(format!("Failed to create HTTP client: {}", e)))?;

        Ok(Self {
            client,
            endpoint: url,
            connected: AtomicBool::new(false),
            request_counter: AtomicU64::new(0),
            stats: Arc::new(parking_lot::Mutex::new(TransportStatistics::default())),
            event_sender: parking_lot::Mutex::new(None),
            queued_events: Arc::new(parking_lot::Mutex::new(Vec::new())),
            session_id: Arc::new(parking_lot::Mutex::new(None)),
            auth_override: Arc::new(parking_lot::RwLock::new(None)),
            protocol_version: Arc::new(parking_lot::RwLock::new("2025-11-25".to_string())),
        })
    }

    /// Create HTTP transport with custom client
    pub fn with_client(endpoint: &str, client: Client) -> McpClientResult<Self> {
        let url = Url::parse(endpoint)
            .map_err(|e| TransportError::ConnectionFailed(format!("Invalid URL: {}", e)))?;

        Ok(Self {
            client,
            endpoint: url,
            connected: AtomicBool::new(false),
            request_counter: AtomicU64::new(0),
            stats: Arc::new(parking_lot::Mutex::new(TransportStatistics::default())),
            event_sender: parking_lot::Mutex::new(None),
            queued_events: Arc::new(parking_lot::Mutex::new(Vec::new())),
            session_id: Arc::new(parking_lot::Mutex::new(None)),
            auth_override: Arc::new(parking_lot::RwLock::new(None)),
            protocol_version: Arc::new(parking_lot::RwLock::new("2025-11-25".to_string())),
        })
    }

    /// Set the session ID to use for subsequent requests
    pub fn set_session_id(&self, session_id: String) {
        debug!("Setting session ID: {}", session_id);
        *self.session_id.lock() = Some(session_id);
    }

    /// Set the negotiated MCP spec version, advertised on the `MCP-Protocol-Version`
    /// header. A stateless version (2026-07-28) also drops any captured session id —
    /// the stateless core has no `Mcp-Session-Id`.
    pub fn set_protocol_version(&self, version: &str) {
        *self.protocol_version.write() = version.to_string();
        if version == "2026-07-28" {
            *self.session_id.lock() = None;
        }
    }

    /// Whether this connection carries `Mcp-Session-Id`. The 2026-07-28
    /// stateless core removed the header, so its absence is not worth a warning
    /// there.
    fn uses_session_header(&self) -> bool {
        *self.protocol_version.read() != "2026-07-28"
    }

    /// Clear the session ID (used during 404 re-initialization)
    pub fn clear_session_id(&self) {
        debug!("Clearing session ID for re-initialization");
        *self.session_id.lock() = None;
    }

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

    /// SEP-2243 request-metadata headers, emitted on 2026-07-28 connections:
    /// mirror the body's `method` into `Mcp-Method` (all requests and
    /// notifications) and `params.name` / `params.uri` into `Mcp-Name` for
    /// `tools/call`, `prompts/get`, and `resources/read`. 2025-11-25
    /// connections keep their original wire shape (no extra headers).
    fn apply_request_metadata_headers(
        &self,
        mut builder: reqwest::RequestBuilder,
        message: &Value,
    ) -> reqwest::RequestBuilder {
        if *self.protocol_version.read() != "2026-07-28" {
            return builder;
        }
        if let Some(method) = message.get("method").and_then(|m| m.as_str()) {
            builder = builder.header("Mcp-Method", method);
            let name = match method {
                "tools/call" | "prompts/get" => {
                    message.pointer("/params/name").and_then(|v| v.as_str())
                }
                "resources/read" => message.pointer("/params/uri").and_then(|v| v.as_str()),
                _ => None,
            };
            if let Some(name) = name {
                // SEP-2243 §Value Encoding: an `Mcp-Name` value that cannot
                // ride as a plain ASCII header value MUST be Base64-sentinel
                // encoded; plain values pass through verbatim.
                #[cfg(any(feature = "client-bilingual", feature = "client-2026-07-28-only"))]
                let header_value = turul_mcp_protocol_2026_07_28::headers::encode_param_value(
                    &Value::String(name.to_string()),
                )
                .unwrap_or_else(|| name.to_string());
                // A 2025-11-25-only build never reaches this 2026 header path
                // at runtime (the protocol-version gate above), so the raw
                // value stands in where the 2026 crate is not linked.
                #[cfg(not(any(feature = "client-bilingual", feature = "client-2026-07-28-only")))]
                let header_value = name.to_string();
                builder = builder.header("Mcp-Name", header_value);
            }
        }
        builder
    }

    /// Apply the live `Authorization` override (if any) to a `RequestBuilder`.
    /// Per-request headers override `default_headers` of the same name in
    /// reqwest, so this transparently supersedes any bearer baked in via
    /// `ConnectionConfig::headers`.
    fn apply_auth_override(&self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        if let Some(value) = self.auth_override.read().as_ref() {
            builder.header(reqwest::header::AUTHORIZATION, value)
        } else {
            builder
        }
    }

    /// Update statistics
    fn update_stats<F>(&self, update_fn: F)
    where
        F: FnOnce(&mut TransportStatistics),
    {
        let mut stats = self.stats.lock();
        update_fn(&mut stats);
    }

    /// Test helper: Process an in-memory byte stream (for testing without network)
    #[doc(hidden)]
    pub async fn test_handle_byte_stream<S, B, E>(&self, stream: S) -> McpClientResult<Value>
    where
        S: Stream<Item = Result<B, E>> + Unpin,
        B: AsRef<[u8]>,
        E: std::error::Error + Send + Sync + 'static,
    {
        self.handle_byte_stream(stream).await
    }

    /// Parse an SSE stream from a POST response.
    /// Extracts JSON-RPC frames from `data:` lines, routes notifications
    /// to event_sender, returns the final result/error frame.
    async fn handle_sse_stream(&self, response: Response) -> McpClientResult<Value> {
        use futures::StreamExt;
        use tokio::io::AsyncBufReadExt;

        let byte_stream = response.bytes_stream();
        let reader = tokio_util::io::StreamReader::new(
            byte_stream.map(|r| r.map_err(std::io::Error::other)),
        );
        let mut lines = tokio::io::BufReader::new(reader).lines();

        let sender_snapshot = self.event_sender.lock().clone();
        parse_sse_lines(
            &mut lines,
            sender_snapshot,
            &self.queued_events,
            &self.stats,
        )
        .await
    }

    /// Test helper: Process an in-memory SSE byte stream
    #[doc(hidden)]
    pub async fn test_handle_sse_stream<S, B, E>(&self, stream: S) -> McpClientResult<Value>
    where
        S: Stream<Item = Result<B, E>> + Unpin,
        B: AsRef<[u8]>,
        E: std::error::Error + Send + Sync + 'static,
    {
        use futures::StreamExt;
        use tokio::io::AsyncBufReadExt;

        // Collect all bytes first (test helper — no streaming needed)
        let mut buffer = Vec::new();
        let mut stream = stream;
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.map_err(|e| TransportError::Http(format!("Stream error: {}", e)))?;
            buffer.extend_from_slice(chunk.as_ref());
        }

        let cursor = std::io::Cursor::new(buffer);
        let mut lines = tokio::io::BufReader::new(cursor).lines();

        let sender_snapshot = self.event_sender.lock().clone();
        parse_sse_lines(
            &mut lines,
            sender_snapshot,
            &self.queued_events,
            &self.stats,
        )
        .await
    }

    /// Streamable HTTP servers answer protocol-level rejections with
    /// `400 Bad Request` and a JSON-RPC error body (UnsupportedProtocolVersion,
    /// HeaderMismatch, MissingRequiredClientCapability). Surface that envelope
    /// to the caller's normal JSON-RPC error classification instead of a
    /// transport failure. Only status 400 with a parseable `error.code`
    /// envelope is rescued — every other status (404 session recovery,
    /// 401/403 auth) keeps transport-error semantics.
    fn rescue_400_jsonrpc_envelope(result: McpClientResult<Value>) -> McpClientResult<Value> {
        match result {
            Err(crate::error::McpClientError::Transport(TransportError::HttpStatus {
                status: 400,
                message,
            })) => match serde_json::from_str::<Value>(&message) {
                Ok(body) if body.pointer("/error/code").is_some() => Ok(body),
                _ => Err(TransportError::HttpStatus {
                    status: 400,
                    message,
                }
                .into()),
            },
            other => other,
        }
    }

    /// Map a non-2xx `(status, body)` to a client error, applying the same
    /// 400-envelope rescue as [`rescue_400_jsonrpc_envelope`] but for callers
    /// (e.g. the streaming path) that return an error directly rather than a
    /// `Value`. A status-400 body carrying a JSON-RPC `error.code` becomes a
    /// `ServerError` so the code reaches normal classification; every other
    /// status stays a transport `HttpStatus` error.
    fn classify_non_2xx(status: u16, message: String) -> crate::error::McpClientError {
        if status == 400
            && let Ok(body) = serde_json::from_str::<Value>(&message)
            && let Some(code) = body.pointer("/error/code").and_then(|c| c.as_i64())
        {
            let msg = body
                .pointer("/error/message")
                .and_then(|v| v.as_str())
                .unwrap_or("server error")
                .to_string();
            let data = body.pointer("/error/data").cloned();
            return crate::error::McpClientError::server_error(code as i32, msg, data);
        }
        TransportError::HttpStatus { status, message }.into()
    }

    /// Handle HTTP response
    async fn handle_response(&self, response: Response) -> McpClientResult<Value> {
        let status = response.status();

        if !status.is_success() {
            let error_text = response
                .text()
                .await
                .unwrap_or_else(|_| "Unknown error".to_string());

            self.update_stats(|stats| {
                stats.errors += 1;
                stats.last_error = Some(format!("HTTP {}: {}", status, error_text));
            });

            return Err(TransportError::HttpStatus {
                status: status.as_u16(),
                message: error_text,
            }
            .into());
        }

        // Capture session ID from response headers if present
        if *self.protocol_version.read() != "2026-07-28"
            && let Some(session_header) = response.headers().get("mcp-session-id")
            && let Ok(session_str) = session_header.to_str()
        {
            debug!("Captured session ID from response: {}", session_str);
            *self.session_id.lock() = Some(session_str.to_owned());
        }

        let content_type = response
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");

        // Always use streaming approach for JSON responses (handles both single and chunked)
        if content_type.contains("application/json") {
            self.handle_json_stream(response).await
        } else if content_type.contains("text/event-stream") {
            self.handle_sse_stream(response).await
        } else {
            Err(TransportError::Http(format!("Unsupported content type: {}", content_type)).into())
        }
    }

    /// Handle JSON stream (works for both single responses and chunked streaming)
    async fn handle_json_stream(&self, response: Response) -> McpClientResult<Value> {
        use futures::StreamExt;

        let stream = response
            .bytes_stream()
            .map(|result| result.map_err(std::io::Error::other));

        self.handle_byte_stream(stream).await
    }

    /// Handle a generic byte stream (for both HTTP responses and in-memory testing)
    async fn handle_byte_stream<S, B, E>(&self, mut stream: S) -> McpClientResult<Value>
    where
        S: Stream<Item = Result<B, E>> + Unpin,
        B: AsRef<[u8]>,
        E: std::error::Error + Send + Sync + 'static,
    {
        use futures::StreamExt;

        // Ensure event channel exists (lazy creation)
        {
            let mut guard = self.event_sender.lock();
            if guard.is_none() {
                let (tx, _rx) = mpsc::unbounded_channel();
                *guard = Some(tx);
            }
        }

        let mut buffer = Vec::new();

        while let Some(chunk) = stream.next().await {
            let chunk = chunk.map_err(|e| TransportError::Http(format!("Stream error: {}", e)))?;
            buffer.extend_from_slice(chunk.as_ref());

            // Use proper incremental JSON parsing that tracks exact bytes consumed
            let mut processed_bytes = 0;
            loop {
                if processed_bytes >= buffer.len() {
                    break;
                }

                // Try to parse a single JSON value from the remaining buffer
                let remaining_buffer = &buffer[processed_bytes..];
                let mut stream = Deserializer::from_slice(remaining_buffer).into_iter::<Value>();

                match stream.next() {
                    Some(Ok(json)) => {
                        // Calculate how many bytes were consumed by this JSON object
                        let consumed_bytes = stream.byte_offset();

                        debug!(
                            "Parsed JSON frame ({} bytes): {}",
                            consumed_bytes,
                            serde_json::to_string(&json).unwrap_or_default()
                        );

                        // Check if this is a final frame (has id + result/error)
                        if let Some(_id) = json.get("id") {
                            if json.get("result").is_some() {
                                self.update_stats(|stats| stats.responses_received += 1);
                                return Ok(json); // Success response
                            } else if json.get("error").is_some() {
                                self.update_stats(|stats| {
                                    stats.errors += 1;
                                    stats.last_error =
                                        json["error"]["message"].as_str().map(|s| s.to_string());
                                });
                                return Ok(json); // Let client layer extract structured error
                            }
                        }

                        // Progress notification or server request - route accordingly
                        if json.get("method").is_some() {
                            let event = if json.get("id").is_some() && !json["id"].is_null() {
                                ServerEvent::Request(json.clone())
                            } else {
                                ServerEvent::Notification(json.clone())
                            };
                            let sender_snapshot = self.event_sender.lock().clone();
                            if let Some(sender) = sender_snapshot {
                                if sender.send(event.clone()).is_err() {
                                    // Channel closed, queue the event for future listeners
                                    debug!("Event channel closed, queuing notification");
                                    self.queued_events.lock().push(event);
                                } else {
                                    debug!(
                                        "Forwarded progress notification via active event channel"
                                    );
                                }
                            } else {
                                // No listener yet, queue the event
                                debug!("No event listener active, queuing progress notification");
                                self.queued_events.lock().push(event);
                            }
                        }

                        // Update processed bytes counter
                        processed_bytes += consumed_bytes;
                    }
                    Some(Err(e)) => {
                        // If we can't parse a complete JSON object, we have an incomplete frame
                        // Wait for more data - the byte_offset approach handles partial JSON correctly
                        debug!(
                            "Incomplete JSON in buffer, waiting for more data. Parse error: {}",
                            e
                        );
                        break;
                    }
                    None => {
                        // No more complete JSON objects to parse, wait for more data
                        break;
                    }
                }
            }

            // Remove processed data from buffer
            if processed_bytes > 0 {
                buffer.drain(..processed_bytes);
            }
        }

        // Stream ended - check if we have a complete response in buffer
        if !buffer.is_empty()
            && let Ok(json) = serde_json::from_slice::<Value>(&buffer)
            && json.get("id").is_some()
            && (json.get("result").is_some() || json.get("error").is_some())
        {
            self.update_stats(|stats| stats.responses_received += 1);
            return Ok(json);
        }

        // If we get here, no final frame was found
        Err(TransportError::Http("Stream ended without final result".to_string()).into())
    }

    /// Handle HTTP response with headers
    async fn handle_response_with_headers(
        &self,
        response: Response,
    ) -> McpClientResult<crate::transport::TransportResponse> {
        let status = response.status();

        if !status.is_success() {
            let error_text = response
                .text()
                .await
                .unwrap_or_else(|_| "Unknown error".to_string());

            self.update_stats(|stats| {
                stats.errors += 1;
                stats.last_error = Some(format!("HTTP {}: {}", status, error_text));
            });

            return Err(TransportError::HttpStatus {
                status: status.as_u16(),
                message: error_text,
            }
            .into());
        }

        // Capture session ID from response headers if present
        if let Some(session_header) = response.headers().get("mcp-session-id")
            && let Ok(session_str) = session_header.to_str()
        {
            debug!("Captured session ID from response headers: {}", session_str);
            *self.session_id.lock() = Some(session_str.to_owned());
        }

        // Extract headers
        let mut headers = std::collections::HashMap::new();
        for (name, value) in response.headers() {
            if let Ok(value_str) = value.to_str() {
                headers.insert(name.to_string(), value_str.to_string());
            }
        }

        let content_type = response
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");

        // Use streaming approach for JSON responses (handles both single and chunked)
        let response_json = if content_type.contains("application/json") {
            self.handle_json_stream(response).await?
        } else if content_type.contains("text/event-stream") {
            self.handle_sse_stream(response).await?
        } else {
            return Err(TransportError::Http(format!(
                "Unsupported content type: {}",
                content_type
            ))
            .into());
        };

        Ok(crate::transport::TransportResponse::new(
            response_json,
            headers,
        ))
    }
}

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

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

    async fn connect(&self) -> McpClientResult<()> {
        debug!(endpoint = %self.endpoint, "Connecting to HTTP endpoint");

        // HTTP is stateless — no persistent connection to establish.
        // Real connectivity is validated by the initialize request that
        // immediately follows in McpClient::connect().
        self.connected.store(true, Ordering::SeqCst);
        info!(endpoint = %self.endpoint, "HTTP transport connected");
        Ok(())
    }

    async fn disconnect(&self) -> McpClientResult<()> {
        debug!("Disconnecting HTTP transport");
        self.connected.store(false, Ordering::SeqCst);

        // Close event sender if exists
        if let Some(sender) = self.event_sender.lock().take() {
            sender.send(ServerEvent::ConnectionLost).ok();
        }

        info!("HTTP transport disconnected");
        Ok(())
    }

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

    async fn send_request(&self, request: Value) -> McpClientResult<Value> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }

        let start_time = Instant::now();

        // Ensure request has an ID
        let mut request = request;
        if request.get("id").is_none() {
            request["id"] = Value::String(self.next_request_id());
        }

        debug!(
            method = request.get("method").and_then(|v| v.as_str()),
            id = request.get("id").and_then(|v| v.as_str()),
            "Sending HTTP request"
        );

        self.update_stats(|stats| stats.requests_sent += 1);

        let mut req_builder = self
            .client
            .post(self.endpoint.clone())
            .header("Content-Type", "application/json")
            .header("Accept", MCP_POST_ACCEPT)
            .header("MCP-Protocol-Version", self.protocol_version.read().clone());

        req_builder = self.apply_request_metadata_headers(req_builder, &request);
        req_builder = self.apply_auth_override(req_builder);

        // Include session ID if we have one
        if let Some(ref session_id) = *self.session_id.lock() {
            debug!("HTTP request using session ID: {}", session_id);
            req_builder = req_builder.header("Mcp-Session-Id", session_id);
        } else if self.uses_session_header() {
            warn!("HTTP request attempted without session ID - server may reject");
        }

        let response = req_builder
            .json(&request)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send request: {}", e)))?;

        let result = Self::rescue_400_jsonrpc_envelope(self.handle_response(response).await)?;

        let elapsed = start_time.elapsed();
        self.update_stats(|stats| {
            let new_avg = if stats.responses_received > 0 {
                (stats.avg_response_time_ms * (stats.responses_received - 1) as f64
                    + elapsed.as_millis() as f64)
                    / stats.responses_received as f64
            } else {
                elapsed.as_millis() as f64
            };
            stats.avg_response_time_ms = new_avg;
        });

        debug!(elapsed_ms = elapsed.as_millis(), "HTTP request completed");

        Ok(result)
    }

    async fn send_request_streaming(
        &self,
        request: Value,
    ) -> McpClientResult<tokio::sync::mpsc::UnboundedReceiver<Value>> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }
        let mut request = request;
        if request.get("id").is_none() {
            request["id"] = Value::String(self.next_request_id());
        }
        let mut req_builder = self
            .client
            .post(self.endpoint.clone())
            .header("Content-Type", "application/json")
            // Every POST to the MCP endpoint MUST advertise both content types,
            // even when the client expects the SSE stream back.
            .header("Accept", MCP_POST_ACCEPT)
            .header("MCP-Protocol-Version", self.protocol_version.read().clone());
        req_builder = self.apply_request_metadata_headers(req_builder, &request);
        req_builder = self.apply_auth_override(req_builder);

        let response = req_builder
            .json(&request)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send request: {}", e)))?;
        if !response.status().is_success() {
            let status = response.status().as_u16();
            let message = response.text().await.unwrap_or_default();
            return Err(Self::classify_non_2xx(status, message));
        }

        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        tokio::spawn(async move {
            use futures::StreamExt;
            let mut stream = response.bytes_stream();
            let mut buffer = String::new();
            while let Some(chunk) = stream.next().await {
                let Ok(chunk) = chunk else { break };
                buffer.push_str(&String::from_utf8_lossy(&chunk));
                while let Some(pos) = buffer.find("\n\n") {
                    let event: String = buffer.drain(..pos + 2).collect();
                    for line in event.lines() {
                        if let Some(data) = line.strip_prefix("data: ")
                            && let Ok(json) = serde_json::from_str::<Value>(data)
                            && tx.send(json).is_err()
                        {
                            // Receiver dropped: the subscription is cancelled.
                            return;
                        }
                    }
                }
            }
        });
        Ok(rx)
    }

    async fn send_request_with_extra_headers(
        &self,
        request: Value,
        extra_headers: &[(String, String)],
    ) -> McpClientResult<Value> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }
        let mut request = request;
        if request.get("id").is_none() {
            request["id"] = Value::String(self.next_request_id());
        }
        self.update_stats(|stats| stats.requests_sent += 1);

        let mut req_builder = self
            .client
            .post(self.endpoint.clone())
            .header("Content-Type", "application/json")
            .header("Accept", MCP_POST_ACCEPT)
            .header("MCP-Protocol-Version", self.protocol_version.read().clone());
        req_builder = self.apply_request_metadata_headers(req_builder, &request);
        for (name, value) in extra_headers {
            req_builder = req_builder.header(name, value);
        }
        req_builder = self.apply_auth_override(req_builder);
        if let Some(ref session_id) = *self.session_id.lock() {
            req_builder = req_builder.header("Mcp-Session-Id", session_id);
        }

        let response = req_builder
            .json(&request)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send request: {}", e)))?;
        Self::rescue_400_jsonrpc_envelope(self.handle_response(response).await)
    }

    async fn send_request_with_headers(
        &self,
        request: Value,
    ) -> McpClientResult<crate::transport::TransportResponse> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }

        let start_time = Instant::now();

        // Ensure request has an ID
        let mut request = request;
        if request.get("id").is_none() {
            request["id"] = Value::String(self.next_request_id());
        }

        debug!(
            method = request.get("method").and_then(|v| v.as_str()),
            id = request.get("id").and_then(|v| v.as_str()),
            "Sending HTTP request with header extraction"
        );

        self.update_stats(|stats| stats.requests_sent += 1);

        let mut req_builder = self
            .client
            .post(self.endpoint.clone())
            .header("Content-Type", "application/json")
            .header("Accept", MCP_POST_ACCEPT)
            .header("MCP-Protocol-Version", self.protocol_version.read().clone());

        req_builder = self.apply_request_metadata_headers(req_builder, &request);
        req_builder = self.apply_auth_override(req_builder);

        // Include session ID if we have one
        if let Some(ref session_id) = *self.session_id.lock() {
            debug!("HTTP request with headers using session ID: {}", session_id);
            req_builder = req_builder.header("Mcp-Session-Id", session_id);
        } else {
            debug!("HTTP request without session ID (expected for initialize)");
        }

        let response = req_builder
            .json(&request)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send request: {}", e)))?;

        let result = self.handle_response_with_headers(response).await?;

        let elapsed = start_time.elapsed();
        self.update_stats(|stats| {
            let new_avg = if stats.responses_received > 0 {
                (stats.avg_response_time_ms * (stats.responses_received - 1) as f64
                    + elapsed.as_millis() as f64)
                    / stats.responses_received as f64
            } else {
                elapsed.as_millis() as f64
            };
            stats.avg_response_time_ms = new_avg;
        });

        debug!(
            elapsed_ms = elapsed.as_millis(),
            "HTTP request with headers completed"
        );

        Ok(result)
    }

    async fn send_notification(&self, notification: Value) -> McpClientResult<()> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }

        debug!(
            method = notification.get("method").and_then(|v| v.as_str()),
            "Sending HTTP notification"
        );

        self.update_stats(|stats| stats.notifications_sent += 1);

        let mut req_builder = self
            .client
            .post(self.endpoint.clone())
            .header("Accept", MCP_POST_ACCEPT)
            .header("Content-Type", "application/json")
            .header("MCP-Protocol-Version", self.protocol_version.read().clone());

        req_builder = self.apply_request_metadata_headers(req_builder, &notification);
        req_builder = self.apply_auth_override(req_builder);

        // Include session ID if we have one
        if let Some(ref session_id) = *self.session_id.lock() {
            debug!("HTTP notification using session ID: {}", session_id);
            req_builder = req_builder.header("Mcp-Session-Id", session_id);
        } else if self.uses_session_header() {
            warn!("HTTP notification attempted without session ID - server may reject");
        }

        let response = req_builder
            .json(&notification)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send notification: {}", e)))?;

        // For notifications, we expect a 204 No Content or similar
        if response.status().is_success() {
            debug!("HTTP notification sent successfully");
            Ok(())
        } else {
            let status = response.status();
            let error_text = response
                .text()
                .await
                .unwrap_or_else(|_| "Unknown error".to_string());
            Err(TransportError::HttpStatus {
                status: status.as_u16(),
                message: error_text,
            }
            .into())
        }
    }

    async fn send_delete(&self, session_id: &str) -> McpClientResult<()> {
        if !self.is_connected() {
            return Err(TransportError::ConnectionFailed("Not connected".to_string()).into());
        }

        info!(
            endpoint = %self.endpoint,
            session_id = session_id,
            "Sending DELETE request for session termination"
        );

        self.update_stats(|stats| stats.requests_sent += 1);
        let start_time = Instant::now();

        let req_builder = self
            .client
            .delete(self.endpoint.clone())
            .header("Content-Type", "application/json")
            .header("MCP-Protocol-Version", self.protocol_version.read().clone())
            .header("Mcp-Session-Id", session_id);

        // Apply live bearer override so callers that rotated the M2M token
        // immediately before disconnect (Transport::update_auth_header) send
        // the DELETE under the *new* bearer, not the one baked into
        // default_headers at transport construction.
        let response = self
            .apply_auth_override(req_builder)
            .send()
            .await
            .map_err(|e| TransportError::Http(format!("Failed to send DELETE request: {}", e)))?;

        let elapsed = start_time.elapsed();
        self.update_stats(|stats| {
            stats.responses_received += 1;
            // Update average response time
            let elapsed_ms = elapsed.as_millis() as f64;
            if stats.responses_received == 1 {
                stats.avg_response_time_ms = elapsed_ms;
            } else {
                stats.avg_response_time_ms = (stats.avg_response_time_ms
                    * (stats.responses_received - 1) as f64
                    + elapsed_ms)
                    / stats.responses_received as f64;
            }
        });

        // DELETE should return 2xx success status
        if response.status().is_success() {
            info!(
                session_id = session_id,
                status = %response.status(),
                elapsed_ms = elapsed.as_millis(),
                "Session DELETE request completed successfully"
            );
            Ok(())
        } else {
            warn!(
                session_id = session_id,
                status = %response.status(),
                elapsed_ms = elapsed.as_millis(),
                "DELETE request failed but continuing with cleanup"
            );
            // Don't fail on DELETE errors - session cleanup should continue locally
            Ok(())
        }
    }

    async fn start_event_listener(&self) -> McpClientResult<EventReceiver> {
        use futures::StreamExt;

        let (tx, rx) = mpsc::unbounded_channel();
        *self.event_sender.lock() = Some(tx.clone());

        // Replay any queued events to the new listener
        let queued_events = {
            let mut queue = self.queued_events.lock();
            let events = queue.clone();
            queue.clear(); // Clear the queue after replaying
            events
        };

        for event in &queued_events {
            if tx.send(event.clone()).is_err() {
                warn!("Failed to replay queued event - channel already closed");
                break;
            }
        }

        if !queued_events.is_empty() {
            debug!(
                "Replayed {} queued events to new listener",
                queued_events.len()
            );
        }

        if !self.is_connected() {
            warn!("Not connected - event listener will be inactive");
            return Ok(rx);
        }

        // 2026-07-28 deleted the GET SSE stream from Streamable HTTP: a
        // conformant server answers the GET with 405. Server-initiated messages
        // arrive on the POST response stream and on `subscriptions/listen`.
        if !self.uses_session_header() {
            debug!("GET SSE listener not started: 2026-07-28 has no GET stream");
            return Ok(rx);
        }

        // Start SSE connection task for GET requests
        let client = self.client.clone();
        let url = self.endpoint.clone();
        let session_id = self.session_id.clone();
        let auth_override = self.auth_override.clone();
        let protocol_version = self.protocol_version.clone();

        info!("Starting SSE event listener for GET requests at: {}", url);

        tokio::spawn(async move {
            loop {
                // GET request with SSE accept header
                let mut request_builder = client
                    .get(url.as_str())
                    .header("Accept", "text/event-stream")
                    .header("MCP-Protocol-Version", protocol_version.read().clone());

                // Apply the live bearer override (if any) so token rotation
                // also reaches the SSE GET. Per-request header overrides
                // default_headers — see Transport::update_auth_header.
                if let Some(value) = auth_override.read().as_ref() {
                    request_builder = request_builder.header(reqwest::header::AUTHORIZATION, value);
                }

                // Snapshot the session ID at request-build time. We need the
                // snapshot (not just the cached value at response time) so the
                // 4xx terminal handler below can compare-and-swap: only clear
                // the cache if it still matches what we sent. A re-initialize
                // running concurrently with an in-flight GET must not have its
                // fresher session ID clobbered.
                let sent_session_id: Option<String> = session_id.lock().clone();
                if let Some(ref current_session_id) = sent_session_id {
                    debug!("SSE request using session ID: {}", current_session_id);
                    request_builder = request_builder.header("Mcp-Session-Id", current_session_id);
                } else {
                    warn!("SSE request attempted without session ID - server may reject");
                }

                let response = match request_builder.send().await {
                    Ok(resp) if resp.status().is_success() => {
                        debug!("SSE connection established, status: {}", resp.status());

                        let content_type = resp
                            .headers()
                            .get("content-type")
                            .and_then(|v| v.to_str().ok())
                            .unwrap_or("");

                        if !content_type.contains("text/event-stream") {
                            warn!(
                                "Server returned content-type '{}' instead of 'text/event-stream'",
                                content_type
                            );
                        }

                        resp
                    }
                    Ok(resp) => {
                        let status = resp.status();
                        warn!("SSE connection failed with status: {}", status);

                        // 4xx is structurally fatal for the SSE GET listener: retrying
                        // the same request unchanged will never succeed. Clear the cached
                        // session header — but only if it still matches what we sent —
                        // so the caller's next initialize POST is sent without a stale
                        // Mcp-Session-Id (mirrors the POST 404 recovery path in
                        // McpClient::send_request_raw). If a re-initialize wrote a
                        // fresher session ID into the cache while our GET was in
                        // flight, leave the new value alone. Then emit the error and
                        // exit; the caller drives
                        // recovery by re-running initialize and restarting the listener.
                        if status.is_client_error() {
                            let mut guard = session_id.lock();
                            if *guard == sent_session_id {
                                *guard = None;
                            }
                            drop(guard);
                            let _ = tx.send(ServerEvent::Error(format!(
                                "SSE GET rejected with HTTP {} — listener exiting",
                                status
                            )));
                            return;
                        }

                        // 5xx and other non-2xx: treat as transient, retry with the
                        // existing static backoff.
                        if tx
                            .send(ServerEvent::Error(format!("HTTP {}", status)))
                            .is_err()
                        {
                            debug!("Event channel closed, stopping SSE listener");
                            return;
                        }
                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                        continue;
                    }
                    Err(e) => {
                        warn!("SSE connection error: {}", e);
                        if tx.send(ServerEvent::Error(e.to_string())).is_err() {
                            debug!("Event channel closed, stopping SSE listener");
                            return;
                        }
                        // Wait before retrying
                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                        continue;
                    }
                };

                let mut stream = response.bytes_stream();
                let mut buffer = String::new();

                while let Some(chunk) = stream.next().await {
                    if tx.is_closed() {
                        debug!("Event receiver dropped, stopping SSE listener");
                        return; // Receiver dropped, exit cleanly
                    }

                    match chunk {
                        Ok(bytes) => {
                            buffer.push_str(&String::from_utf8_lossy(&bytes));

                            // Parse SSE events (data: {json}\n\n)
                            while let Some(end) = buffer.find("\n\n") {
                                let event_text = buffer[..end].to_string();
                                buffer.drain(..end + 2);

                                // Parse SSE event fields
                                let mut event_type = None;
                                let mut data = String::new();
                                let mut id = None;

                                for line in event_text.lines() {
                                    if let Some(event_value) = line.strip_prefix("event: ") {
                                        event_type = Some(event_value.to_string());
                                    } else if let Some(data_value) = line.strip_prefix("data: ") {
                                        if !data.is_empty() {
                                            data.push('\n');
                                        }
                                        data.push_str(data_value);
                                    } else if let Some(id_value) = line.strip_prefix("id: ") {
                                        id = Some(id_value.to_string());
                                    }
                                }

                                if !data.is_empty() {
                                    // Try to parse as JSON
                                    match serde_json::from_str::<Value>(&data) {
                                        Ok(json) => {
                                            debug!(
                                                "Received SSE event: type={:?}, id={:?}, data={}",
                                                event_type, id, data
                                            );

                                            // Determine event type based on JSON content
                                            // Check method+id first: server-initiated requests
                                            // have both (e.g. sampling/createMessage)
                                            if json.get("method").is_some()
                                                && json.get("id").is_some()
                                            {
                                                // Server-initiated request needing response
                                                if tx.send(ServerEvent::Request(json)).is_err() {
                                                    debug!("Event channel closed during send");
                                                    return;
                                                }
                                            } else if json.get("method").is_some() {
                                                // Notification (method only, no id)
                                                if tx.send(ServerEvent::Notification(json)).is_err()
                                                {
                                                    debug!("Event channel closed during send");
                                                    return;
                                                }
                                            } else if json.get("id").is_some() {
                                                // Response to a pending client request (no method)
                                                if tx.send(ServerEvent::Response(json)).is_err() {
                                                    debug!("Event channel closed during send");
                                                    return;
                                                }
                                            } else {
                                                // Other data
                                                if tx.send(ServerEvent::Notification(json)).is_err()
                                                {
                                                    debug!("Event channel closed during send");
                                                    return;
                                                }
                                            }
                                        }
                                        Err(e) => {
                                            warn!(
                                                "Failed to parse SSE data as JSON: {} - data: {}",
                                                e, data
                                            );
                                        }
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            warn!("SSE stream error: {}", e);
                            break;
                        }
                    }
                }

                // Connection lost, send event and retry
                info!("SSE connection lost, attempting to reconnect...");
                if tx.send(ServerEvent::ConnectionLost).is_err() {
                    debug!("Event channel closed, stopping SSE listener");
                    return;
                }

                // Wait before retrying
                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            }
        });

        info!("SSE event listener started successfully");
        Ok(rx)
    }

    fn connection_info(&self) -> ConnectionInfo {
        ConnectionInfo {
            transport_type: self.transport_type(),
            endpoint: self.endpoint.to_string(),
            connected: self.is_connected(),
            capabilities: self.capabilities(),
            metadata: serde_json::json!({
                "scheme": self.endpoint.scheme(),
                "host": self.endpoint.host_str(),
                "port": self.endpoint.port(),
                "path": self.endpoint.path()
            }),
        }
    }

    async fn health_check(&self) -> McpClientResult<bool> {
        if !self.is_connected() {
            return Ok(false);
        }

        // Send a ping request
        let ping_request = serde_json::json!({
            "jsonrpc": "2.0",
            "id": "health_check",
            "method": "ping",
            "params": {}
        });

        match self.send_request(ping_request).await {
            Ok(_) => Ok(true),
            Err(e) => {
                warn!(error = %e, "Health check failed");
                Ok(false)
            }
        }
    }

    fn set_session_id(&self, session_id: String) {
        debug!("HttpTransport: Setting session ID: {}", session_id);
        *self.session_id.lock() = Some(session_id);
    }

    fn clear_session_id(&self) {
        debug!("HttpTransport: Clearing session ID for re-initialization");
        *self.session_id.lock() = None;
    }

    async fn update_auth_header(&self, value: Option<String>) {
        // RwLock write contention is irrelevant — rotation is rare, reads are
        // common and held only for the duration of building one RequestBuilder.
        *self.auth_override.write() = value;
        debug!("HttpTransport: Authorization override updated");
    }

    fn set_protocol_version(&self, version: &str) {
        HttpTransport::set_protocol_version(self, version);
    }

    fn statistics(&self) -> TransportStatistics {
        self.stats.lock().clone()
    }
}

/// Parse SSE lines, route server requests/notifications, return final response frame.
/// Extracted from handle_sse_stream for testability.
async fn parse_sse_lines<R: tokio::io::AsyncBufRead + Unpin>(
    lines: &mut tokio::io::Lines<R>,
    event_sender: Option<mpsc::UnboundedSender<ServerEvent>>,
    queued_events: &Arc<parking_lot::Mutex<Vec<ServerEvent>>>,
    stats: &Arc<parking_lot::Mutex<TransportStatistics>>,
) -> McpClientResult<Value> {
    while let Ok(Some(line)) = lines.next_line().await {
        let data = line
            .strip_prefix("data: ")
            .or_else(|| line.strip_prefix("data:"));
        let Some(data) = data else {
            continue;
        };
        let Ok(json) = serde_json::from_str::<Value>(data) else {
            continue;
        };
        // Final response frame — has id + result or error
        if json.get("id").is_some() && (json.get("result").is_some() || json.get("error").is_some())
        {
            stats.lock().responses_received += 1;
            return Ok(json);
        }
        // Server request or notification — route to event channel
        if json.get("method").is_some() {
            let event = if json.get("id").is_some() && !json["id"].is_null() {
                ServerEvent::Request(json)
            } else {
                ServerEvent::Notification(json)
            };
            if let Some(ref sender) = event_sender {
                if sender.send(event.clone()).is_err() {
                    queued_events.lock().push(event);
                }
            } else {
                queued_events.lock().push(event);
            }
        }
    }

    Err(TransportError::Http("SSE stream ended without final result".to_string()).into())
}

// Use parking_lot for better performance
use parking_lot;

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

    #[test]
    fn test_http_transport_creation() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();
        assert_eq!(transport.transport_type(), TransportType::Http);
        assert!(!transport.is_connected());
    }

    #[test]
    fn test_invalid_url() {
        let result = HttpTransport::new("invalid-url");
        assert!(result.is_err());
    }

    #[test]
    fn test_invalid_scheme() {
        let result = HttpTransport::new("ftp://localhost:8080/mcp");
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_connect_sets_connected_flag() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();
        assert!(!transport.is_connected());
        transport.connect().await.unwrap();
        assert!(transport.is_connected());
    }

    #[tokio::test]
    async fn test_connection_info() {
        let transport = HttpTransport::new("https://example.com:8443/mcp/endpoint").unwrap();
        let info = transport.connection_info();

        assert_eq!(info.transport_type, TransportType::Http);
        assert_eq!(info.endpoint, "https://example.com:8443/mcp/endpoint");
        assert!(!info.connected);

        let metadata = info.metadata.as_object().unwrap();
        assert_eq!(metadata["scheme"], "https");
        assert_eq!(metadata["host"], "example.com");
        assert_eq!(metadata["port"], 8443);
        assert_eq!(metadata["path"], "/mcp/endpoint");
    }

    #[test]
    fn test_accept_header_contains_both_media_types() {
        assert!(MCP_POST_ACCEPT.contains("application/json"));
        assert!(MCP_POST_ACCEPT.contains("text/event-stream"));
    }

    #[test]
    fn test_request_id_generation() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();

        let id1 = transport.next_request_id();
        let id2 = transport.next_request_id();

        assert_ne!(id1, id2);
        assert!(id1.starts_with("req_"));
        assert!(id2.starts_with("req_"));
    }

    /// Test helper to create an in-memory byte stream for testing
    fn create_test_stream(
        data: Vec<Vec<u8>>,
    ) -> impl Stream<Item = Result<Vec<u8>, std::io::Error>> + Unpin {
        futures::stream::iter(data.into_iter().map(Ok))
    }

    #[tokio::test]
    async fn test_handle_byte_stream_single_response() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();

        // Create a single JSON response
        let response_json = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "result": {
                "tools": []
            }
        });
        let response_bytes = serde_json::to_vec(&response_json).unwrap();

        // Create stream with single chunk
        let stream = create_test_stream(vec![response_bytes]);

        // Process the stream
        let result = transport.handle_byte_stream(stream).await.unwrap();

        assert_eq!(result["jsonrpc"], "2.0");
        assert_eq!(result["id"], 1);
        assert!(result.get("result").is_some());
    }

    #[tokio::test]
    async fn test_handle_byte_stream_chunked_response() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();

        // Create a response that will be split across chunks
        let response_json = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "result": {
                "tools": [
                    {"name": "calculator", "description": "A calculator tool"}
                ]
            }
        });
        let response_bytes = serde_json::to_vec(&response_json).unwrap();

        // Split the response into multiple chunks to test streaming
        let chunk1 = response_bytes[..20].to_vec();
        let chunk2 = response_bytes[20..40].to_vec();
        let chunk3 = response_bytes[40..].to_vec();

        let stream = create_test_stream(vec![chunk1, chunk2, chunk3]);

        // Process the stream
        let result = transport.handle_byte_stream(stream).await.unwrap();

        assert_eq!(result["jsonrpc"], "2.0");
        assert_eq!(result["id"], 2);
        assert!(result.get("result").is_some());
    }

    #[tokio::test]
    async fn test_handle_byte_stream_with_progress_notifications() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();

        // Create progress notification followed by final response
        let progress1 = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": {
                "progressToken": "test_progress",
                "progress": 50,
                "total": 100
            }
        });

        let progress2 = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": {
                "progressToken": "test_progress",
                "progress": 100,
                "total": 100
            }
        });

        let final_response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 3,
            "result": {
                "success": true
            }
        });

        // Create separate chunks for each frame
        let progress1_bytes = serde_json::to_vec(&progress1).unwrap();
        let progress2_bytes = serde_json::to_vec(&progress2).unwrap();
        let final_bytes = serde_json::to_vec(&final_response).unwrap();

        let stream = create_test_stream(vec![progress1_bytes, progress2_bytes, final_bytes]);

        // Start event listener to capture progress notifications
        let mut events = transport.start_event_listener().await.unwrap();

        // Process the stream (should return final response)
        let result = transport.handle_byte_stream(stream).await.unwrap();

        // Verify final response
        assert_eq!(result["jsonrpc"], "2.0");
        assert_eq!(result["id"], 3);
        assert_eq!(result["result"]["success"], true);

        // Verify progress notifications were captured (with timeout to avoid hanging)
        use tokio::time::{Duration, timeout};

        let event1 = timeout(Duration::from_millis(100), events.recv()).await;
        assert!(event1.is_ok() && event1.unwrap().is_some());

        let event2 = timeout(Duration::from_millis(100), events.recv()).await;
        assert!(event2.is_ok() && event2.unwrap().is_some());
    }

    #[tokio::test]
    async fn test_progress_event_queue_replay() {
        let transport = HttpTransport::new("http://localhost:8080/mcp").unwrap();

        // Create progress notifications that arrive BEFORE starting event listener
        let progress1 = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": {
                "progressToken": "early_progress",
                "progress": 25,
                "total": 100
            }
        });

        let progress2 = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": {
                "progressToken": "early_progress",
                "progress": 75,
                "total": 100
            }
        });

        let final_response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 4,
            "result": {
                "queued_events_test": true
            }
        });

        // Create separate chunks for each frame
        let progress1_bytes = serde_json::to_vec(&progress1).unwrap();
        let progress2_bytes = serde_json::to_vec(&progress2).unwrap();
        let final_bytes = serde_json::to_vec(&final_response).unwrap();

        let stream = create_test_stream(vec![progress1_bytes, progress2_bytes, final_bytes]);

        // Process stream WITHOUT starting event listener first
        // This should queue the progress notifications
        let result = transport.handle_byte_stream(stream).await.unwrap();

        // Verify final response
        assert_eq!(result["jsonrpc"], "2.0");
        assert_eq!(result["id"], 4);
        assert_eq!(result["result"]["queued_events_test"], true);

        // NOW start event listener - should replay queued events
        let mut events = transport.start_event_listener().await.unwrap();

        // Verify queued events are replayed (with timeout to avoid hanging)
        use tokio::time::{Duration, timeout};

        let event1 = timeout(Duration::from_millis(100), events.recv()).await;
        assert!(
            event1.is_ok() && event1.unwrap().is_some(),
            "Should replay first queued event"
        );

        let event2 = timeout(Duration::from_millis(100), events.recv()).await;
        assert!(
            event2.is_ok() && event2.unwrap().is_some(),
            "Should replay second queued event"
        );

        // Verify no more events (queue should be empty after replay)
        let no_more_events = timeout(Duration::from_millis(50), events.recv()).await;
        assert!(
            no_more_events.is_err(),
            "Should not have any more events after replay"
        );
    }

    #[tokio::test]
    async fn test_handle_sse_post_response() {
        let sse_body: &[u8] = b"event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{\"tools\":[]}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_body.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;
        assert!(
            result.is_ok(),
            "SSE POST response should parse successfully"
        );
        let json = result.unwrap();
        assert_eq!(json["id"], "req_0");
        assert!(json["result"]["tools"].is_array());
    }

    #[tokio::test]
    async fn test_handle_sse_post_response_error() {
        let sse_body: &[u8] = b"data: {\"jsonrpc\":\"2.0\",\"id\":\"req_1\",\"error\":{\"code\":-32600,\"message\":\"Invalid\"}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_body.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;
        assert!(result.is_ok());
        let json = result.unwrap();
        assert_eq!(json["error"]["code"], -32600);
    }

    #[tokio::test]
    async fn test_handle_sse_post_no_final_frame() {
        let sse_body: &[u8] =
            b"data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\"}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_body.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;
        assert!(result.is_err(), "Should fail when no final frame");
    }

    #[test]
    fn test_with_config_custom_headers() {
        use std::collections::HashMap;
        let mut headers = HashMap::new();
        headers.insert("Authorization".to_string(), "Bearer test-token".to_string());

        let config = crate::config::ConnectionConfig {
            headers: Some(headers),
            ..Default::default()
        };

        let transport = HttpTransport::with_config("http://localhost:9999/mcp", &config);
        assert!(transport.is_ok());
    }

    #[test]
    fn test_with_config_no_redirects() {
        let config = crate::config::ConnectionConfig {
            follow_redirects: false,
            ..Default::default()
        };

        let transport = HttpTransport::with_config("http://localhost:9999/mcp", &config);
        assert!(transport.is_ok());
    }

    #[test]
    fn test_with_config_default() {
        let config = crate::config::ConnectionConfig::default();
        let transport = HttpTransport::with_config("http://localhost:9999/mcp", &config);
        assert!(transport.is_ok());
    }

    #[test]
    fn test_http_transport_advertises_server_events() {
        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let caps = transport.capabilities();
        assert!(
            caps.server_events,
            "HttpTransport must advertise server_events"
        );
        assert!(!caps.bidirectional, "HTTP is not bidirectional");
    }

    #[tokio::test]
    async fn test_server_request_routed_as_request_event() {
        // A server-initiated request has both method AND id
        let server_request =
            br#"{"jsonrpc":"2.0","id":"srv-1","method":"sampling/createMessage","params":{}}"#;
        let stream = create_test_stream(vec![server_request.to_vec()]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        // This will fail to find a final response frame (the server request is not a response),
        // but the frame should be queued as a Request event
        let _ = transport.test_handle_byte_stream(stream).await;

        let events = transport.queued_events.lock();
        assert!(!events.is_empty(), "Server request should be queued");
        match &events[0] {
            crate::transport::ServerEvent::Request(val) => {
                assert_eq!(val["method"], "sampling/createMessage");
                assert_eq!(val["id"], "srv-1");
            }
            other => panic!("Expected ServerEvent::Request, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_jsonrpc_error_preserved_through_transport() {
        let error_response = br#"{"jsonrpc":"2.0","id":"req_0","error":{"code":-32602,"message":"Invalid params","data":{"detail":"missing field"}}}"#;
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(error_response.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_byte_stream(stream).await;

        assert!(
            result.is_ok(),
            "JSON-RPC error should pass through transport as Ok(Value)"
        );
        let json = result.unwrap();
        assert_eq!(json["error"]["code"], -32602);
        assert_eq!(json["error"]["message"], "Invalid params");
        assert_eq!(json["error"]["data"]["detail"], "missing field");
    }

    #[tokio::test]
    async fn test_sse_post_with_server_request_routed_correctly() {
        let sse_data = b"data: {\"jsonrpc\":\"2.0\",\"id\":\"srv-99\",\"method\":\"sampling/createMessage\",\"params\":{}}\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{\"tools\":[]}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_data.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;

        assert!(result.is_ok());
        let json = result.unwrap();
        assert_eq!(
            json["id"], "req_0",
            "Should return the final response frame"
        );

        let events = transport.queued_events.lock();
        assert_eq!(events.len(), 1, "Should have exactly one queued event");
        match &events[0] {
            ServerEvent::Request(val) => {
                assert_eq!(val["id"], "srv-99");
                assert_eq!(val["method"], "sampling/createMessage");
            }
            other => panic!("Expected ServerEvent::Request, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_sse_post_with_notification_routed_correctly() {
        let sse_data = b"data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{\"progress\":50}}\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{\"tools\":[]}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_data.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;

        assert!(result.is_ok());

        let events = transport.queued_events.lock();
        assert_eq!(events.len(), 1);
        match &events[0] {
            ServerEvent::Notification(val) => {
                assert_eq!(val["method"], "notifications/progress");
            }
            other => panic!("Expected ServerEvent::Notification, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_sse_data_field_without_space_after_colon() {
        // SSE spec allows "data:{json}" without space after colon
        let sse_body: &[u8] = b"data:{\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_body.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;
        assert!(
            result.is_ok(),
            "SSE data field without space should be parseable: {:?}",
            result.err()
        );
    }

    /// "Per the SSE specification, any line beginning with a colon is a
    /// comment that carries no event data; clients must ignore such lines and
    /// must not treat them as malformed input" (basic/transports/streamable-http).
    /// A `:`-prefixed line matches neither the `data:` nor `data: ` prefix
    /// `parse_sse_lines` checks for, so it falls into the `continue` branch —
    /// this proves a comment line surrounded by real frames neither breaks
    /// parsing nor is misrouted as a notification/response.
    #[tokio::test]
    async fn test_sse_colon_comment_line_is_ignored() {
        let sse_data = b": this is a comment, ignore me\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{\"progress\":50}}\n: another comment\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{\"tools\":[]}}\n\n";
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(sse_data.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_sse_stream(stream).await;

        assert!(
            result.is_ok(),
            "comment lines must not be treated as malformed input: {:?}",
            result.err()
        );
        assert_eq!(
            result.unwrap()["id"],
            "req_0",
            "final result frame must still be parsed past the comment lines"
        );

        let events = transport.queued_events.lock();
        assert_eq!(
            events.len(),
            1,
            "exactly one notification event, comment lines produce none"
        );
        match &events[0] {
            ServerEvent::Notification(val) => {
                assert_eq!(val["method"], "notifications/progress");
            }
            other => panic!("Expected ServerEvent::Notification, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_sse_post_no_duplicate_event_delivery() {
        use tokio::io::AsyncBufReadExt;

        let (tx, mut rx) = mpsc::unbounded_channel();
        let queued = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let stats = Arc::new(parking_lot::Mutex::new(TransportStatistics::default()));

        // SSE with one notification then final result
        let sse_data = b"data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{}}\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"req_0\",\"result\":{}}\n\n";
        let cursor = std::io::Cursor::new(sse_data.to_vec());
        let mut lines = tokio::io::BufReader::new(cursor).lines();

        let sender = Some(tx);
        let _ = parse_sse_lines(&mut lines, sender, &queued, &stats).await;

        // Check: event should be in EITHER the channel OR queued_events, not both
        let channel_event = rx.try_recv().ok();
        let queued_events = queued.lock().clone();

        let total = (if channel_event.is_some() { 1 } else { 0 }) + queued_events.len();
        assert_eq!(
            total,
            1,
            "Event should be delivered exactly once, not duplicated. Channel: {}, Queued: {}",
            channel_event.is_some(),
            queued_events.len()
        );
    }

    #[tokio::test]
    async fn test_byte_stream_request_vs_notification_discrimination() {
        // Server request (has method + id) followed by notification (method only) followed by final result
        let frames = br#"{"jsonrpc":"2.0","id":"srv-1","method":"sampling/createMessage","params":{}}{"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}{"jsonrpc":"2.0","id":"req_0","result":{"tools":[]}}"#;
        let stream = futures::stream::iter(vec![Ok::<_, std::io::Error>(frames.to_vec())]);

        let transport = HttpTransport::new("http://localhost:9999/mcp").unwrap();
        let result = transport.test_handle_byte_stream(stream).await;

        assert!(result.is_ok());

        let events = transport.queued_events.lock();
        assert_eq!(
            events.len(),
            2,
            "Should have both request and notification events"
        );

        // First event: server request
        match &events[0] {
            ServerEvent::Request(val) => {
                assert_eq!(val["id"], "srv-1");
                assert_eq!(val["method"], "sampling/createMessage");
            }
            other => panic!(
                "Expected ServerEvent::Request as first event, got {:?}",
                other
            ),
        }

        // Second event: notification
        match &events[1] {
            ServerEvent::Notification(val) => {
                assert_eq!(val["method"], "notifications/progress");
            }
            other => panic!(
                "Expected ServerEvent::Notification as second event, got {:?}",
                other
            ),
        }
    }
}