//! Streamable HTTP transport (MCP 2025-03-26 spec).
//!
//! Why a dedicated module: the streamable HTTP wire shape mixes plain
//! JSON-RPC POST responses with SSE-framed streaming responses on the
//! same endpoint. Keeping that logic in one place lets the protocol
//! [`Client`](crate::protocol::Client) consume it through the channel
//! pair contract without caring about HTTP semantics.
//!
//! Responsibilities:
//!
//! - POST each outbound JSON-RPC envelope to the configured endpoint.
//! - Capture `Mcp-Session-Id` from the first response and replay it on
//!   every subsequent request.
//! - Inject static headers (including a `bearer_token` shortcut that
//!   sets `Authorization: Bearer <token>`).
//! - When the POST response is `text/event-stream`, parse the stream and
//!   feed JSON-RPC envelopes back into the inbound channel.
//! - Open a long-lived SSE GET against the endpoint so server-initiated
//!   notifications and requests reach the Client.
//! - Reconnect with exponential backoff (50ms, 100ms, 200ms, 400ms,
//!   give up) when the SSE stream drops with a transient failure.
//!
//! The transport spawns two tokio tasks: a writer that drains the
//! outbound channel and posts messages, and a reader that owns the SSE
//! GET. Both tasks exit when their channel ends close, which is the
//! same shutdown contract the protocol Client uses.

mod auth_hint;
mod post;
mod wire;

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use bytes::Bytes;
use futures_util::StreamExt;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT};
use reqwest::{Client as HttpClient, Method};
use serde_json::Value;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::sleep;

use auth_hint::auth_hint;

use crate::protocol::{JsonRpcMessage, Transport};
use crate::transport::header_capture::HeaderCapture;
use crate::transport::sse_parser::SseParser;

/// Default per-request timeout when the caller does not pin one.
pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// Default connect timeout when the caller does not pin one.
pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
/// Header the spec uses to carry the session identifier.
pub const SESSION_HEADER: &str = "Mcp-Session-Id";
/// SEP-2243 routing header carrying the JSON-RPC method name (for
/// example `tools/call`, `resources/read`, `prompts/get`). Emitted on
/// every request in stateless mode so a server can route without
/// parsing the body.
pub const METHOD_HEADER: &str = "Mcp-Method";
/// SEP-2243 routing header carrying the target name when the method
/// has one: the tool name for `tools/call`, the resource URI for
/// `resources/read`, the prompt name for `prompts/get`. Skipped for
/// methods that have no name target. Stateless mode only.
pub const NAME_HEADER: &str = "Mcp-Name";
/// JSON-RPC error code for a transport-level HTTP rejection (4xx) that has no
/// JSON-RPC body. Sits in the implementation-defined server-error range
/// (-32000 to -32099) so it never collides with a code the peer might send.
const HTTP_TRANSPORT_ERROR_CODE: i64 = -32010;

/// Backoff schedule for transient failures (5xx, network error).
///
/// Picked from the  acceptance criteria: 50ms, 100ms, 200ms,
/// 400ms, then surface the error to the caller. Exposed as a constant
/// so tests can introspect it without re-reading the source.
pub const BACKOFF_STEPS_MS: &[u64] = &[50, 100, 200, 400];

/// Construction parameters for [`StreamableHttp::connect`].
///
/// Held as a struct (rather than a long positional argument list) so
/// future knobs (proxy, TLS overrides) can land without breaking the
/// builder shape.
#[derive(Debug, Clone)]
pub struct StreamableHttpConfig {
    /// Endpoint URL. POSTed for client requests, opened with GET for the
    /// server-to-client SSE stream.
    pub url: String,
    /// Static headers injected on every request. Authorization-style
    /// headers can live here; the [`bearer_token`](Self::bearer_token)
    /// shortcut is purely a convenience.
    pub headers: HashMap<String, String>,
    /// Optional bearer token. When set, the transport sends
    /// `Authorization: Bearer <token>` on every request. Mutually
    /// exclusive in intent with an `Authorization` entry in
    /// [`headers`](Self::headers); when both are present the explicit
    /// header wins (last-write semantics inside reqwest).
    pub bearer_token: Option<String>,
    /// Per-request timeout (POST). Default 30s.
    pub request_timeout: Duration,
    /// TCP and TLS connect timeout. Default 30s.
    pub connect_timeout: Duration,
    /// HTTP / HTTPS proxy policy. The default empty config preserves
    /// reqwest's automatic env-var pickup; explicit fields override.
    /// See [`mcptest_core::network::ProxyConfig`](crate::network::ProxyConfig).
    pub proxy: crate::network::ProxyConfig,
    /// Optional mTLS client identity. Applied at client-build
    /// time as a `reqwest::Identity`; composes with `bearer_token` and
    /// static `headers`.
    pub mtls: Option<crate::auth::MtlsConfig>,
    /// Optional AWS SigV4 request signer. When set, every POST
    /// and SSE GET is signed per request with fresh credentials and a
    /// fresh timestamp.
    pub sigv4: Option<crate::auth::SigV4Config>,
    /// Optional Web Bot Auth request signer. When set, every
    /// POST and SSE GET carries RFC 9421 `Signature` headers.
    pub web_bot_auth: Option<crate::auth::WebBotAuthConfig>,
    /// Routing mode. `Session` (default) captures `Mcp-Session-Id`
    /// and replays it. `Stateless` (2026-07-28, SEP-2575) drops
    /// session-id and attaches a per-request `_meta` block instead.
    pub mode: TransportMode,
    /// `_meta` value spliced into every outbound request when in
    /// stateless mode. Build with
    /// [`crate::protocol::build_request_meta`]. Ignored otherwise.
    pub stateless_meta: Option<Value>,
}

/// Wire-level transport routing mode (SEP-2575).
///
/// The default `Session` matches the pre-2026-07-28 spec shapes: the
/// transport captures `Mcp-Session-Id` from the first response and
/// replays it on every later request. `Stateless` is the 2026-07-28
/// shape: no session header, and every outbound request carries a
/// `_meta` block with protocol version + client info + capabilities so
/// any server instance can satisfy the call.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum TransportMode {
    /// Session-based: capture and replay `Mcp-Session-Id` (pre-2026-07-28).
    #[default]
    Session,
    /// Stateless: drop session-id, attach `_meta` to every request.
    Stateless,
}

impl Default for StreamableHttpConfig {
    fn default() -> Self {
        Self {
            url: String::new(),
            headers: HashMap::new(),
            bearer_token: None,
            request_timeout: DEFAULT_REQUEST_TIMEOUT,
            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
            proxy: crate::network::ProxyConfig::default(),
            mtls: None,
            sigv4: None,
            web_bot_auth: None,
            mode: TransportMode::default(),
            stateless_meta: None,
        }
    }
}

/// Errors raised while constructing a [`StreamableHttp`] transport.
///
/// Once connected the transport surfaces every wire failure through the
/// inbound channel by closing it, so these errors only cover the
/// upfront setup step (URL parsing, reqwest client construction).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum StreamableHttpError {
    /// The URL was empty or could not be parsed.
    #[error("invalid endpoint URL `{0}`")]
    InvalidUrl(String),
    /// A configured header name or value was not valid for `reqwest`.
    #[error("invalid header `{name}`: {reason}")]
    InvalidHeader {
        /// Header name as supplied by the caller.
        name: String,
        /// Underlying reason from the HTTP stack.
        reason: String,
    },
    /// `reqwest` rejected the client construction (TLS, DNS, etc.).
    #[error("reqwest client construction failed: {0}")]
    Reqwest(#[from] reqwest::Error),
    /// Proxy URL or combination rejected at builder time.
    #[error(transparent)]
    Proxy(#[from] crate::network::ProxyError),
    /// A client-auth scheme (mTLS) failed to build at connect time.
    #[error("client auth setup failed: {0}")]
    Auth(#[from] crate::auth::AuthError),
}

/// Live streamable HTTP transport handle.
///
/// Holds the spawned writer and reader tasks plus the channels the
/// protocol Client drives. Drop the handle to abort both tasks; the
/// usual route is to call [`StreamableHttp::shutdown`] explicitly.
pub struct StreamableHttp {
    writer: JoinHandle<()>,
    reader: JoinHandle<()>,
    header_capture: HeaderCapture,
}

/// Shared mutable state between the writer and reader tasks.
///
/// Only the session id needs cross-task coordination so far. Wrapped in
/// a `std::sync::Mutex` because writes are infrequent (one update on
/// first POST) and tokio's Mutex is overkill for that workload.
#[derive(Default)]
struct SharedState {
    session_id: Mutex<Option<String>>,
    /// Routing mode for the writer + response dispatch. Decided at
    /// connect time; never changes during the transport's lifetime.
    mode: TransportMode,
    /// `_meta` block spliced into every outbound request when running
    /// in `Stateless` mode. None in session mode.
    stateless_meta: Option<Value>,
    /// Per-request response header capture (WOR-1287). The writer task
    /// records into it; the executor takes entries out via the clone
    /// exposed on [`StreamableHttp::header_capture`].
    header_capture: HeaderCapture,
}

impl SharedState {
    fn get_session(&self) -> Option<String> {
        // poisoned mutex is impossible here without a panic; treat that
        // as no session yet to avoid `unwrap()`.
        self.session_id.lock().ok().and_then(|g| g.clone())
    }

    fn set_session(&self, value: String) {
        if let Ok(mut guard) = self.session_id.lock() {
            *guard = Some(value);
        }
    }
}

impl StreamableHttp {
    /// Open the transport and return a [`Transport`] plus the handle.
    ///
    /// The transport spawns its writer and reader tasks on the current
    /// tokio runtime. Both tasks shut down cleanly when the protocol
    /// Client drops its end of the channels.
    pub fn connect(config: StreamableHttpConfig) -> Result<(Transport, Self), StreamableHttpError> {
        if config.url.trim().is_empty() {
            return Err(StreamableHttpError::InvalidUrl(config.url));
        }
        let mut base_headers = HeaderMap::new();
        for (name, value) in &config.headers {
            let hname = HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
                StreamableHttpError::InvalidHeader {
                    name: name.clone(),
                    reason: e.to_string(),
                }
            })?;
            let hval =
                HeaderValue::from_str(value).map_err(|e| StreamableHttpError::InvalidHeader {
                    name: name.clone(),
                    reason: e.to_string(),
                })?;
            base_headers.insert(hname, hval);
        }
        if let Some(token) = &config.bearer_token {
            let v = HeaderValue::from_str(&format!("Bearer {token}")).map_err(|e| {
                StreamableHttpError::InvalidHeader {
                    name: "Authorization".into(),
                    reason: e.to_string(),
                }
            })?;
            base_headers.insert(reqwest::header::AUTHORIZATION, v);
        }

        let mut builder = HttpClient::builder()
            .connect_timeout(config.connect_timeout)
            .timeout(config.request_timeout)
            .default_headers(base_headers);
        builder = config.proxy.apply(builder)?;
        builder = apply_mtls(builder, config.mtls.as_ref())?;
        let http = builder.build()?;

        tracing::debug!(
            target: "mcptest_core::transport::streamable_http",
            url = %config.url,
            "streamable-http transport connected"
        );

        let (outbound_tx, outbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (inbound_tx, inbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();

        let shared = Arc::new(SharedState {
            session_id: Mutex::new(None),
            mode: config.mode,
            stateless_meta: config.stateless_meta.clone(),
            header_capture: HeaderCapture::default(),
        });
        let header_capture = shared.header_capture.clone();
        let signer = Arc::new(crate::auth::RequestSigner::from_configs(
            config.sigv4.clone(),
            config.web_bot_auth.clone(),
        ));

        let writer = spawn_writer(
            http.clone(),
            config.url.clone(),
            Arc::clone(&shared),
            Arc::clone(&signer),
            outbound_rx,
            inbound_tx.clone(),
        );
        let reader = spawn_reader(http, config.url, Arc::clone(&shared), signer, inbound_tx);

        let transport = Transport::new(outbound_tx, inbound_rx);
        Ok((
            transport,
            Self {
                writer,
                reader,
                header_capture,
            },
        ))
    }

    /// Clone of the per-request response header capture channel
    /// (WOR-1287). The connector grabs this before the handle moves into
    /// its shutdown slot, so the executor can take each call's headers
    /// for the `response_headers` budget gate.
    pub fn header_capture(&self) -> HeaderCapture {
        self.header_capture.clone()
    }

    /// Abort the writer and reader tasks.
    ///
    /// Idempotent and safe to call from anywhere. Tasks may have already
    /// exited on their own when the protocol Client closed the channels.
    pub async fn shutdown(self) {
        self.writer.abort();
        self.reader.abort();
        // Wait for both to finish so the runtime can drop their state
        // before returning. JoinErrors after `abort` carry no caller
        // useful information, so we deliberately swallow them.
        let _ = self.writer.await;
        let _ = self.reader.await;
    }
}

/// Apply an optional mTLS client identity to the reqwest builder.
///
/// A no-op when `mtls` is absent or disabled. Otherwise it attaches the
/// client identity and, when a CA bundle is configured, an extra root
/// certificate. Composes with bearer tokens and static headers.
fn apply_mtls(
    builder: reqwest::ClientBuilder,
    mtls: Option<&crate::auth::MtlsConfig>,
) -> Result<reqwest::ClientBuilder, StreamableHttpError> {
    let Some(cfg) = mtls else {
        return Ok(builder);
    };
    if !cfg.is_enabled() {
        return Ok(builder);
    }
    let mut builder = builder.identity(crate::auth::identity(cfg)?);
    if let Some(root) = crate::auth::root_certificate(cfg)? {
        builder = builder.add_root_certificate(root);
    }
    Ok(builder)
}

/// Drain the outbound channel, POST each envelope, and route responses.
///
/// On a 503 or network failure the writer retries per [`BACKOFF_STEPS_MS`]
/// before giving up on the envelope. Responses come back as either
/// `application/json` (one response, dispatched into the inbound
/// channel) or `text/event-stream` (multi-message stream, parsed via
/// `SseParser` and dispatched as each event arrives).
fn spawn_writer(
    http: HttpClient,
    url: String,
    shared: Arc<SharedState>,
    signer: Arc<Option<crate::auth::RequestSigner>>,
    mut outbound_rx: mpsc::UnboundedReceiver<JsonRpcMessage>,
    inbound_tx: mpsc::UnboundedSender<JsonRpcMessage>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Some(msg) = outbound_rx.recv().await {
            if post::post_with_retry(
                &http,
                &url,
                &shared,
                signer.as_ref().as_ref(),
                &msg,
                &inbound_tx,
            )
            .await
            .is_err()
            {
                // Stream is poisoned. Close the inbound channel so the
                // protocol Client sees `TransportClosed`. Returning ends
                // the loop, which drops `inbound_tx`.
                return;
            }
        }
    })
}

/// Run the per-request signer (SigV4 / Web Bot Auth) and attach the
/// resulting headers. A no-op when no signer is configured. A signing
/// failure is fatal: it surfaces as a non-transient [`PostError`] so the
/// writer poisons the stream rather than retrying a doomed request.
fn apply_signature(
    mut req: reqwest::RequestBuilder,
    signer: Option<&crate::auth::RequestSigner>,
    method: &str,
    url: &str,
    body: &[u8],
) -> Result<reqwest::RequestBuilder, post::PostError> {
    let Some(signer) = signer else {
        return Ok(req);
    };
    let headers = signer.sign(method, url, body).map_err(|e| {
        tracing::error!(
            target: "mcptest_core::transport::streamable_http",
            error = %e,
            "request signing failed"
        );
        post::PostError {
            transient: false,
            retry_after: None,
        }
    })?;
    for h in headers {
        req = req.header(h.name, h.value);
    }
    Ok(req)
}

/// Sign the long-lived SSE GET (empty body). A signing failure is fatal
/// for the reader loop, surfaced as [`SseGetError::Signing`].
fn sign_get(
    mut req: reqwest::RequestBuilder,
    signer: Option<&crate::auth::RequestSigner>,
    url: &str,
) -> Result<reqwest::RequestBuilder, SseGetError> {
    let Some(signer) = signer else {
        return Ok(req);
    };
    let headers = signer.sign("GET", url, &[]).map_err(|e| {
        tracing::error!(
            target: "mcptest_core::transport::streamable_http",
            error = %e,
            "SSE GET signing failed"
        );
        SseGetError::Signing
    })?;
    for h in headers {
        req = req.header(h.name, h.value);
    }
    Ok(req)
}

/// Long-lived SSE reader. Opens a GET request to the endpoint and
/// streams server-pushed envelopes into the inbound channel.
///
/// On a transient failure we reconnect with the same backoff schedule
/// as the writer. The reader cooperates with shutdown by checking the
/// inbound channel before each retry; once the Client drops its end
/// the `send` calls below fail and the loop exits.
fn spawn_reader(
    http: HttpClient,
    url: String,
    shared: Arc<SharedState>,
    signer: Arc<Option<crate::auth::RequestSigner>>,
    inbound_tx: mpsc::UnboundedSender<JsonRpcMessage>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut attempt = 0usize;
        let mut last_event_id: Option<String> = None;
        loop {
            if inbound_tx.is_closed() {
                return;
            }
            match run_sse_get(
                &http,
                &url,
                &shared,
                signer.as_ref().as_ref(),
                &inbound_tx,
                last_event_id.as_deref(),
            )
            .await
            {
                Ok(final_id) => {
                    // Stream ended cleanly. Reset backoff and reconnect.
                    last_event_id = final_id;
                    attempt = 0;
                }
                Err(SseGetError::Signing) => {
                    // Fatal: a signing failure will recur. Stop retrying.
                    return;
                }
                Err(SseGetError::Reqwest(_e)) => {
                    if attempt >= BACKOFF_STEPS_MS.len() {
                        return;
                    }
                    sleep(Duration::from_millis(BACKOFF_STEPS_MS[attempt])).await;
                    attempt += 1;
                }
            }
        }
    })
}

/// Failure modes of one SSE GET: a transient wire error the reader backs
/// off on, or a fatal signing error that will recur and so stops the loop.
enum SseGetError {
    /// Wire failure (network, non-2xx). Retryable.
    Reqwest(reqwest::Error),
    /// Per-request signing failed. Fatal.
    Signing,
}

impl From<reqwest::Error> for SseGetError {
    fn from(e: reqwest::Error) -> Self {
        SseGetError::Reqwest(e)
    }
}

/// Open one SSE GET. Returns the most recent event id seen so the next
/// reconnect can replay with `Last-Event-Id`.
async fn run_sse_get(
    http: &HttpClient,
    url: &str,
    shared: &SharedState,
    signer: Option<&crate::auth::RequestSigner>,
    inbound_tx: &mpsc::UnboundedSender<JsonRpcMessage>,
    last_event_id: Option<&str>,
) -> Result<Option<String>, SseGetError> {
    let mut req = http
        .request(Method::GET, url)
        .header(ACCEPT, "text/event-stream");
    // Skip the session header in stateless mode, same rule as the
    // POST path.
    if matches!(shared.mode, TransportMode::Session) {
        if let Some(sid) = shared.get_session() {
            req = req.header(SESSION_HEADER, sid);
        }
    }
    if let Some(id) = last_event_id {
        req = req.header("Last-Event-Id", id);
    }
    req = sign_get(req, signer, url)?;
    let resp = req.send().await?;
    if !resp.status().is_success() {
        // No retry budget here: surface as error so the caller waits.
        return Err(resp.error_for_status().unwrap_err().into());
    }
    let mut stream = resp.bytes_stream();
    let mut parser = SseParser::new();
    let mut latest_id: Option<String> = last_event_id.map(|s| s.to_owned());
    while let Some(chunk) = stream.next().await {
        let bytes: Bytes = chunk?;
        parser.feed(&bytes);
        for ev in parser.take_ready() {
            if let Some(id) = &ev.id {
                latest_id = Some(id.clone());
            }
            if ev.data.is_empty() {
                continue;
            }
            post::dispatch_json_bytes(ev.data.as_bytes(), inbound_tx);
        }
    }
    Ok(latest_id)
}

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

    #[test]
    fn backoff_schedule_matches_spec() {
        assert_eq!(BACKOFF_STEPS_MS, &[50, 100, 200, 400]);
    }

    #[test]
    fn retry_wait_uses_retry_after_seconds_when_present() {
        let d = post::retry_wait(0, Some(2));
        assert_eq!(d, Duration::from_secs(2));
    }

    #[test]
    fn retry_wait_falls_back_to_backoff_steps() {
        assert_eq!(post::retry_wait(0, None), Duration::from_millis(50));
        assert_eq!(post::retry_wait(3, None), Duration::from_millis(400));
        assert_eq!(post::retry_wait(99, None), Duration::from_millis(400));
    }

    #[test]
    fn config_default_uses_thirty_second_timeouts() {
        let c = StreamableHttpConfig::default();
        assert_eq!(c.request_timeout, Duration::from_secs(30));
        assert_eq!(c.connect_timeout, Duration::from_secs(30));
    }

    #[test]
    fn dispatch_json_bytes_parses_array_response() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let body = br#"[
            {"jsonrpc":"2.0","id":1,"result":{"ok":true}},
            {"jsonrpc":"2.0","method":"notice","params":{}}
        ]"#;
        post::dispatch_json_bytes(body, &tx);
        let m1 = rx.try_recv().expect("first envelope");
        let m2 = rx.try_recv().expect("second envelope");
        assert!(m1.result.is_some());
        assert_eq!(m2.method.as_deref(), Some("notice"));
    }

    #[test]
    fn shared_state_reports_no_session_initially() {
        let s = SharedState::default();
        assert!(s.get_session().is_none());
        s.set_session("abc".into());
        assert_eq!(s.get_session().as_deref(), Some("abc"));
    }

    // === Transport mode routing (SEP-2575) ===

    fn session_state() -> SharedState {
        SharedState {
            session_id: Mutex::new(None),
            mode: TransportMode::Session,
            stateless_meta: None,
            header_capture: HeaderCapture::default(),
        }
    }

    fn stateless_state(meta: Value) -> SharedState {
        SharedState {
            session_id: Mutex::new(None),
            mode: TransportMode::Stateless,
            stateless_meta: Some(meta),
            header_capture: HeaderCapture::default(),
        }
    }

    #[test]
    fn config_default_mode_is_session_for_backward_compat() {
        // Pin the default so a caller that does not opt into stateless
        // mode keeps the pre-2026-07-28 wire behavior unchanged.
        let c = StreamableHttpConfig::default();
        assert_eq!(c.mode, TransportMode::Session);
        assert!(c.stateless_meta.is_none());
    }

    #[test]
    fn build_outbound_body_session_mode_does_not_inject_meta() {
        // The session-based path must serialize exactly the message we
        // were handed; injecting `_meta` against a pre-2026-07-28
        // server would be wrong.
        let msg = JsonRpcMessage::request(
            crate::protocol::JsonRpcId::Number(1),
            "tools/call".to_string(),
            serde_json::json!({"name": "echo", "arguments": {"q": "hi"}}),
        );
        let body = wire::build_outbound_body(&msg, &session_state());
        let parsed: Value = serde_json::from_slice(&body).expect("body parses");
        assert!(
            parsed.get("params").and_then(|p| p.get("_meta")).is_none(),
            "session mode must not inject `_meta`"
        );
    }

    #[test]
    fn build_outbound_body_stateless_mode_splices_meta_into_params() {
        let meta = serde_json::json!({
            "protocolVersion": "2026-07-28",
            "clientInfo": {"name": "mcptest", "version": "1.0.0"},
            "capabilities": {}
        });
        let msg = JsonRpcMessage::request(
            crate::protocol::JsonRpcId::Number(1),
            "tools/call".to_string(),
            serde_json::json!({"name": "echo", "arguments": {"q": "hi"}}),
        );
        let body = wire::build_outbound_body(&msg, &stateless_state(meta.clone()));
        let parsed: Value = serde_json::from_slice(&body).expect("body parses");
        // Original args survive.
        assert_eq!(parsed["params"]["name"], serde_json::json!("echo"));
        // `_meta` lands under params and matches the configured value.
        assert_eq!(parsed["params"]["_meta"], meta);
    }

    #[test]
    fn build_outbound_body_stateless_mode_leaves_envelope_alone_when_params_is_missing() {
        // A notification with no params field stays as-is; the spec
        // says `_meta` rides on a `params` object, not on a scalar
        // or absent field.
        let mut value = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "ping"
        });
        let msg: JsonRpcMessage = serde_json::from_value(value.take()).unwrap();
        let state = stateless_state(serde_json::json!({"protocolVersion": "2026-07-28"}));
        let body = wire::build_outbound_body(&msg, &state);
        let parsed: Value = serde_json::from_slice(&body).expect("body parses");
        assert!(parsed.get("params").is_none() || parsed["params"].get("_meta").is_none());
    }

    // === SEP-2243 Mcp-Method / Mcp-Name routing headers ===

    #[test]
    fn name_from_params_pulls_tool_call_name() {
        let p = serde_json::json!({"name": "echo", "arguments": {"q": "hi"}});
        assert_eq!(wire::name_from_params("tools/call", Some(&p)), Some("echo"));
    }

    #[test]
    fn name_from_params_pulls_resources_read_uri() {
        let p = serde_json::json!({"uri": "file:///etc/hosts"});
        assert_eq!(
            wire::name_from_params("resources/read", Some(&p)),
            Some("file:///etc/hosts")
        );
    }

    #[test]
    fn name_from_params_pulls_prompts_get_name() {
        let p = serde_json::json!({"name": "summarize"});
        assert_eq!(
            wire::name_from_params("prompts/get", Some(&p)),
            Some("summarize")
        );
    }

    #[test]
    fn name_from_params_returns_none_for_methods_without_a_named_target() {
        let p = serde_json::json!({});
        assert!(wire::name_from_params("tools/list", Some(&p)).is_none());
        assert!(wire::name_from_params("initialize", Some(&p)).is_none());
        assert!(wire::name_from_params("ping", Some(&p)).is_none());
    }

    #[test]
    fn name_from_params_returns_none_when_params_is_missing() {
        assert!(wire::name_from_params("tools/call", None).is_none());
    }

    #[test]
    fn name_from_params_returns_none_when_the_expected_field_is_not_a_string() {
        let p = serde_json::json!({"name": 42});
        assert!(wire::name_from_params("tools/call", Some(&p)).is_none());
    }

    #[test]
    fn name_from_params_returns_none_for_tools_call_without_a_name_field() {
        let p = serde_json::json!({"arguments": {"q": "hi"}});
        assert!(wire::name_from_params("tools/call", Some(&p)).is_none());
    }

    #[test]
    fn splice_meta_replaces_an_existing_meta_key() {
        // A pre-existing `_meta` in params is overwritten by the
        // transport's value, so the caller cannot accidentally bake in
        // a stale trace context or version stamp.
        let envelope = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "echo",
                "_meta": {"protocolVersion": "2025-06-18"}
            }
        });
        let new_meta = serde_json::json!({"protocolVersion": "2026-07-28"});
        let result = wire::splice_meta(envelope, &new_meta);
        assert_eq!(result["params"]["_meta"], new_meta);
    }

    #[tokio::test]
    async fn connect_with_empty_url_errors() {
        let cfg = StreamableHttpConfig {
            url: "".into(),
            ..Default::default()
        };
        match StreamableHttp::connect(cfg) {
            Err(StreamableHttpError::InvalidUrl(_)) => {}
            Err(other) => panic!("expected InvalidUrl, got {other}"),
            Ok(_) => panic!("expected error for empty URL"),
        }
    }
}
