mermaid-cli 0.30.0

//! Streamable HTTP JSON-RPC 2.0 transport for MCP servers (spec 2025-11-25).
//!
//! Every client message is POSTed to the single MCP endpoint; the server
//! answers a request with either a plain JSON body or an SSE stream that
//! eventually carries the response. Session state (`MCP-Session-Id`) and the
//! negotiated protocol version (`MCP-Protocol-Version`) ride as headers on
//! every subsequent request. Shutdown is an HTTP DELETE of the session.
//!
//! OAuth: when the server uses it, every request carries the bearer token
//! from `mermaid mcp login` (see `super::oauth`).
//!
//! Out of scope (v1): the GET server-listening stream, and the
//! deprecated two-endpoint HTTP+SSE transport. SSE resumption (GET +
//! `Last-Event-ID`) is implemented only for the SEP-1699 polling case: a
//! server that closes the connection mid-request after assigning event IDs.

use anyhow::{Context, Result, anyhow, bail};
use futures::StreamExt;
use reqwest::StatusCode;
use reqwest::header::{ACCEPT, AUTHORIZATION, HeaderMap, HeaderName, HeaderValue};
use serde_json::{Value, json};
use std::sync::RwLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use tokio::time::{Duration, timeout};

use super::oauth::OAuthSession;
use super::transport::{
    JsonRpcError, REQUEST_TIMEOUT_SECS, RequestTimeout, extract_jsonrpc_result, is_response,
    parse_response_id,
};
use mermaid_domain::{McpServerConfig, TransportKind};
use mermaid_model::utils::{CredentialStore, HostClass, classify_host, drain_sse_events};

/// TCP connect budget. Separate from the response budget: a dead host should
/// fail in seconds, not eat the whole 30s control-call window.
const CONNECT_TIMEOUT_SECS: u64 = 10;

/// Budget for a POSTed notification/response (202-class round-trip) — no
/// server compute is expected, only delivery.
const NOTIFICATION_TIMEOUT_SECS: u64 = 10;

/// Budget for the session DELETE at shutdown. Best-effort; shutdown never
/// blocks on a slow server.
const DELETE_TIMEOUT_SECS: u64 = 5;

/// Cap on SEP-1699 SSE reconnects for a single request. The outer per-request
/// timeout still bounds total wall time; this stops a server that keeps
/// closing the stream without ever delivering the response.
const MAX_SSE_RECONNECTS: u32 = 5;

/// Max bytes of an error-response body echoed into an error message. Redacted
/// before use — bodies can quote request headers.
const ERROR_BODY_SNIPPET_BYTES: usize = 200;

// `HeaderName::from_static` requires lowercase. HTTP header names are
// case-insensitive on the wire; the spec currently spells these
// `MCP-Session-Id` / `MCP-Protocol-Version` / `Last-Event-ID`.
const SESSION_HEADER: HeaderName = HeaderName::from_static("mcp-session-id");
const PROTOCOL_VERSION_HEADER: HeaderName = HeaderName::from_static("mcp-protocol-version");
const LAST_EVENT_ID_HEADER: HeaderName = HeaderName::from_static("last-event-id");
/// 2026-07-28: the JSON-RPC method, mirrored for intermediaries.
const METHOD_HEADER: HeaderName = HeaderName::from_static("mcp-method");

/// Accept values: POSTs must advertise both response shapes; SSE resume GETs
/// only ever get a stream.
const ACCEPT_POST: &str = "application/json, text/event-stream";
const ACCEPT_SSE: &str = "text/event-stream";

/// Streamable HTTP transport for a single remote MCP server.
///
/// # Concurrency model
///
/// Each request is an independent HTTP round-trip correlated by JSON-RPC id
/// (`next_id` is atomic), so concurrent callers don't serialize the way stdio
/// writes do. `session_id` / `protocol_version` are `RwLock`s written once
/// early in the session (initialize) and read on every request; the locks are
/// never held across an `.await`.
pub(super) struct HttpTransport {
    client: reqwest::Client,
    url: reqwest::Url,
    /// Literal config headers, validated at construction.
    static_headers: HeaderMap,
    /// Header name -> env var name, resolved per request so a rotated secret
    /// is picked up without restart. A missing var is skipped.
    env_headers: Vec<(HeaderName, String)>,
    /// `MCP-Session-Id` captured from response headers; echoed on every
    /// subsequent request once set.
    session_id: RwLock<Option<HeaderValue>>,
    /// Negotiated protocol version, set from the initialize result; sent as
    /// `MCP-Protocol-Version` on everything after.
    protocol_version: RwLock<Option<HeaderValue>>,
    /// Monotonic request ID counter.
    next_id: AtomicU64,
    /// OAuth bearer tokens from `mermaid mcp login`. `None` when the config
    /// sends its own `Authorization` header.
    oauth: Option<OAuthSession>,
    /// Speaking the 2026-07-28 protocol: `Mcp-Method` on every request, no
    /// session, no SSE resumption.
    modern: AtomicBool,
}

impl HttpTransport {
    /// Build a transport for server `name` with `config` (which must be
    /// url-shaped). Fails fast on an invalid url/scheme or an unparseable
    /// header NAME — error messages never include header values, which are
    /// secrets. OAuth tokens come from the OS keyring.
    pub fn new(name: &str, config: &McpServerConfig) -> Result<Self> {
        Self::with_store(name, config, None)
    }

    /// [`Self::new`] with the token store injected (`None` = OS keyring).
    pub(super) fn with_store(
        name: &str,
        config: &McpServerConfig,
        store: Option<std::sync::Arc<dyn CredentialStore>>,
    ) -> Result<Self> {
        // Re-validate rather than trust the caller: enforces url-presence and
        // the https-or-loopback scheme rule on every construction path.
        if config.transport_kind()? != TransportKind::Http {
            bail!("MCP server config is not url-shaped");
        }
        let url_str = config
            .url
            .as_deref()
            .expect("transport_kind checked url presence");
        let url = reqwest::Url::parse(url_str)
            .map_err(|e| anyhow!("invalid MCP server url '{url_str}': {e}"))?;

        check_ip_literal(&url, config.allow_private_network)?;

        let mut static_headers = HeaderMap::new();
        for (name, value) in &config.headers {
            let n = HeaderName::from_bytes(name.as_bytes())
                .map_err(|_| anyhow!("invalid MCP header name '{name}'"))?;
            // Never echo the value: it is a secret.
            let v = HeaderValue::from_str(value)
                .map_err(|_| anyhow!("invalid value for MCP header '{name}'"))?;
            static_headers.insert(n, v);
        }
        let mut env_headers = Vec::new();
        for (name, var) in &config.env_headers {
            let n = HeaderName::from_bytes(name.as_bytes())
                .map_err(|_| anyhow!("invalid MCP header name '{name}'"))?;
            env_headers.push((n, var.clone()));
        }

        let client = vetted_client(config.allow_private_network)?;

        let oauth = OAuthSession::for_server(name, config, store, client.clone());
        Ok(Self {
            client,
            url,
            static_headers,
            env_headers,
            session_id: RwLock::new(None),
            protocol_version: RwLock::new(None),
            next_id: AtomicU64::new(1),
            oauth,
            modern: AtomicBool::new(false),
        })
    }

    /// Send a JSON-RPC request and wait for the response under the default
    /// control-call timeout.
    pub async fn send_request(&self, method: &str, params: Value) -> Result<Value> {
        self.send_request_with_timeout(method, params, REQUEST_TIMEOUT_SECS)
            .await
    }

    /// Like [`Self::send_request`], but with an explicit budget bounding the
    /// WHOLE round-trip: POST, SSE drain, and any SEP-1699 reconnects.
    pub async fn send_request_with_timeout(
        &self,
        method: &str,
        params: Value,
        response_timeout_secs: u64,
    ) -> Result<Value> {
        self.send_request_with_headers(method, params, response_timeout_secs, &HeaderMap::new())
            .await
    }

    /// Like [`Self::send_request_with_timeout`], with extra headers on the
    /// POST (the 2026-07-28 `Mcp-Name` and `Mcp-Param-*` headers).
    pub async fn send_request_with_headers(
        &self,
        method: &str,
        params: Value,
        response_timeout_secs: u64,
        extra: &HeaderMap,
    ) -> Result<Value> {
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let request = json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": method,
            "params": params,
        });
        timeout(
            Duration::from_secs(response_timeout_secs),
            self.request_roundtrip(method, id, &request, extra),
        )
        .await
        .map_err(|_| RequestTimeout {
            method: method.to_string(),
            secs: response_timeout_secs,
        })?
    }

    /// Send a JSON-RPC notification. The server answers a client notification
    /// (or response) with 202 Accepted; any 2xx is tolerated.
    pub async fn send_notification(&self, method: &str, params: Value) -> Result<()> {
        let notification = json!({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
        });
        self.post_message(&notification)
            .await
            .with_context(|| format!("MCP notification failed (method: {method})"))
    }

    /// Enter (`Some(version)`) or leave (`None`) the 2026-07-28 shape. In it,
    /// every request carries `MCP-Protocol-Version: <version>` and
    /// `Mcp-Method`, and no session id is kept.
    pub fn set_modern(&self, version: Option<&str>) {
        self.modern.store(version.is_some(), Ordering::Release);
        *self
            .protocol_version
            .write()
            .expect("mcp protocol_version lock poisoned") =
            version.and_then(|v| HeaderValue::from_str(v).ok());
        *self
            .session_id
            .write()
            .expect("mcp session_id lock poisoned") = None;
    }

    fn is_modern(&self) -> bool {
        self.modern.load(Ordering::Acquire)
    }

    /// Record the protocol version negotiated during `initialize`; sent as the
    /// `MCP-Protocol-Version` header on every subsequent request.
    pub fn set_protocol_version(&self, version: &str) {
        match HeaderValue::from_str(version) {
            Ok(v) => {
                *self
                    .protocol_version
                    .write()
                    .expect("mcp protocol_version lock poisoned") = Some(v);
            },
            // A version that can't be a header value can't go on the wire;
            // requests then omit the header (server assumes 2025-03-26).
            Err(_) => tracing::warn!(
                "MCP: server negotiated protocol version {:?} is not a valid header value; \
                 omitting MCP-Protocol-Version",
                version
            ),
        }
    }

    /// End the session: DELETE with the session id, bounded to 5s. A server
    /// MAY respond 405 (it does not allow client-initiated termination) —
    /// tolerated, like every other failure here. No session id = nothing to do.
    pub async fn shutdown(&self) {
        if self
            .session_id
            .read()
            .expect("mcp session_id lock poisoned")
            .is_none()
        {
            return;
        }
        let Ok(headers) = self.request_headers(None) else {
            return;
        };
        let result = timeout(
            Duration::from_secs(DELETE_TIMEOUT_SECS),
            self.send_authed(
                headers,
                |h| self.client.delete(self.url.clone()).headers(h),
                || "MCP session DELETE failed".to_string(),
            ),
        )
        .await;
        match result {
            Ok(Ok(resp)) if resp.status() == StatusCode::METHOD_NOT_ALLOWED => {
                tracing::debug!("MCP: server does not allow client session termination (405)");
            },
            Ok(Err(e)) => tracing::debug!("MCP: session DELETE failed: {:#}", e),
            Err(_) => tracing::debug!("MCP: session DELETE timed out"),
            Ok(Ok(_)) => {},
        }
    }

    /// One request round-trip: POST, then dispatch on the response
    /// Content-Type (plain JSON object vs SSE stream).
    async fn request_roundtrip(
        &self,
        method: &str,
        id: u64,
        request: &Value,
        extra: &HeaderMap,
    ) -> Result<Value> {
        let mut headers = self.request_headers(Some(ACCEPT_POST))?;
        for (name, value) in extra {
            headers.insert(name, value.clone());
        }
        if self.is_modern() {
            headers.insert(
                METHOD_HEADER,
                HeaderValue::from_str(method)
                    .map_err(|_| anyhow!("MCP method '{method}' is not a valid header value"))?,
            );
        }
        let response = self
            .send_authed(
                headers,
                |h| self.client.post(self.url.clone()).headers(h).json(request),
                || format!("MCP HTTP request failed (method: {method})"),
            )
            .await?;
        self.capture_session(response.headers());

        let status = response.status();
        if status == StatusCode::ACCEPTED {
            // 202 is the notification/response acknowledgement; a request
            // needs an actual JSON-RPC response.
            bail!("MCP server returned 202 Accepted to a request (method: {method})");
        }
        if !status.is_success() {
            return Err(self.status_error(method, status, response).await);
        }

        let content_type = response
            .headers()
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("")
            .to_ascii_lowercase();
        if content_type.starts_with("text/event-stream") {
            let msg = self.drain_sse_for_response(response, method, id).await?;
            extract_jsonrpc_result(msg)
        } else if content_type.starts_with("application/json") {
            let msg: Value = response
                .json()
                .await
                .with_context(|| format!("MCP response was not valid JSON (method: {method})"))?;
            if !is_response(&msg) || msg.get("id").and_then(parse_response_id) != Some(id) {
                bail!("MCP JSON response did not answer request {id} (method: {method})");
            }
            extract_jsonrpc_result(msg)
        } else {
            bail!(
                "MCP server returned unsupported Content-Type '{content_type}' \
                 (method: {method}; expected application/json or text/event-stream)"
            );
        }
    }

    /// Drain an SSE response stream until the response to request `id`
    /// arrives. Handles SEP-1699 polling: if the server assigned event IDs and
    /// the connection closes before our response, honor any `retry` delay and
    /// resume via GET + `Last-Event-ID` (resumption is always a GET, even for
    /// a POST-originated stream).
    async fn drain_sse_for_response(
        &self,
        response: reqwest::Response,
        method: &str,
        id: u64,
    ) -> Result<Value> {
        let mut meta = SseMeta::default();
        let mut response = response;
        let mut reconnects = 0u32;
        loop {
            if let Some(msg) = self.drain_one_stream(response, id, &mut meta).await? {
                return Ok(msg);
            }
            // Stream ended without our response. Without event IDs there is
            // nothing to resume from — the server just dropped the request.
            // 2026-07-28 has no resumption at all: the request is lost.
            let Some(last_id) = meta.last_event_id.clone().filter(|_| !self.is_modern()) else {
                bail!("MCP SSE stream ended without a response (method: {method})");
            };
            reconnects += 1;
            if reconnects > MAX_SSE_RECONNECTS {
                bail!(
                    "MCP SSE stream did not deliver a response after {MAX_SSE_RECONNECTS} \
                     reconnects (method: {method})"
                );
            }
            // The client MUST respect a server-provided retry delay.
            if let Some(ms) = meta.retry_ms {
                tokio::time::sleep(Duration::from_millis(ms)).await;
            }
            let mut headers = self.request_headers(Some(ACCEPT_SSE))?;
            headers.insert(
                LAST_EVENT_ID_HEADER,
                HeaderValue::from_str(&last_id)
                    .map_err(|_| anyhow!("MCP SSE event id is not a valid header value"))?,
            );
            let resumed = self
                .send_authed(
                    headers,
                    |h| self.client.get(self.url.clone()).headers(h),
                    || format!("MCP SSE resume failed (method: {method})"),
                )
                .await?;
            self.capture_session(resumed.headers());
            if !resumed.status().is_success() {
                let status = resumed.status();
                return Err(self.status_error(method, status, resumed).await);
            }
            response = resumed;
        }
    }

    /// Drain one SSE HTTP response. Returns `Ok(Some(msg))` when the response
    /// to `id` arrives, `Ok(None)` on a disconnect that may be resumable.
    async fn drain_one_stream(
        &self,
        response: reqwest::Response,
        id: u64,
        meta: &mut SseMeta,
    ) -> Result<Option<Value>> {
        meta.start_stream();
        let mut stream = response.bytes_stream();
        let mut buf: Vec<u8> = Vec::new();
        while let Some(chunk) = stream.next().await {
            let chunk = match chunk {
                Ok(c) => c,
                Err(e) => {
                    // A mid-stream error after the server assigned event IDs
                    // is indistinguishable from a deliberate SEP-1699 close;
                    // let the caller resume. Without IDs it is fatal.
                    if meta.last_event_id.is_some() {
                        tracing::debug!("MCP: SSE stream broke, will resume: {}", e);
                        return Ok(None);
                    }
                    return Err(anyhow!(e)).context("MCP SSE stream read failed");
                },
            };
            meta.feed(&chunk);
            buf.extend_from_slice(&chunk);
            for payload in drain_sse_events(&mut buf) {
                let Ok(msg) = serde_json::from_str::<Value>(&payload) else {
                    tracing::warn!("MCP: unparseable SSE event payload");
                    continue;
                };
                if is_response(&msg) {
                    if msg.get("id").and_then(parse_response_id) == Some(id) {
                        return Ok(Some(msg));
                    }
                    // A response to some other request on this stream —
                    // nothing of ours to complete.
                    continue;
                }
                if msg.get("method").is_some() {
                    match msg.get("id") {
                        // Server-initiated request: it blocks on a reply, so
                        // POST one back rather than stalling the server.
                        Some(rid) if !rid.is_null() => {
                            let rid = rid.clone();
                            self.answer_server_request(&msg, &rid).await;
                        },
                        // Notification: no reply expected. Ignore it.
                        _ => tracing::trace!(
                            method = msg.get("method").and_then(|m| m.as_str()).unwrap_or(""),
                            "MCP: ignoring server notification on SSE stream"
                        ),
                    }
                }
            }
        }
        Ok(None)
    }

    /// Answer a server-initiated request arriving on an SSE stream: `ping`
    /// gets a proper empty result (spec lifecycle utility); anything else gets
    /// JSON-RPC -32601. Best-effort — a failed reply is the server's stall,
    /// not ours.
    async fn answer_server_request(&self, msg: &Value, rid: &Value) {
        let method = msg.get("method").and_then(|m| m.as_str()).unwrap_or("");
        let reply = if method == "ping" {
            json!({ "jsonrpc": "2.0", "id": rid, "result": {} })
        } else {
            json!({
                "jsonrpc": "2.0",
                "id": rid,
                "error": { "code": -32601, "message": "method not supported" },
            })
        };
        if let Err(e) = self.post_message(&reply).await {
            tracing::debug!(
                "MCP: failed to answer server-initiated request '{}': {}",
                method,
                e
            );
        }
    }

    /// POST a message that expects no JSON-RPC response (a notification or a
    /// response of ours). Any 2xx acknowledges it; non-2xx = server rejected.
    async fn post_message(&self, message: &Value) -> Result<()> {
        let fut = async {
            let response = self
                .send_authed(
                    self.request_headers(Some(ACCEPT_POST))?,
                    |h| self.client.post(self.url.clone()).headers(h).json(message),
                    || "MCP HTTP post failed".to_string(),
                )
                .await?;
            self.capture_session(response.headers());
            let status = response.status();
            if !status.is_success() {
                return Err(self.status_error("(notification)", status, response).await);
            }
            Ok(())
        };
        timeout(Duration::from_secs(NOTIFICATION_TIMEOUT_SECS), fut)
            .await
            .map_err(|_| anyhow!("MCP notification timed out after {NOTIFICATION_TIMEOUT_SECS}s"))?
    }

    /// Send one request built by `build`, with the OAuth bearer token when
    /// this server uses OAuth. On 401 it retries once with a refreshed token;
    /// with none to be had, or on a 403 `insufficient_scope`, the error is
    /// [`AuthRequired`](super::oauth::AuthRequired), naming the login command.
    /// `what` labels transport failures.
    async fn send_authed(
        &self,
        headers: HeaderMap,
        build: impl Fn(HeaderMap) -> reqwest::RequestBuilder,
        what: impl Fn() -> String,
    ) -> Result<reqwest::Response> {
        let Some(oauth) = &self.oauth else {
            return build(headers).send().await.with_context(what);
        };
        let with_bearer = |bearer: Option<&HeaderValue>| {
            let mut h = headers.clone();
            if let Some(b) = bearer {
                h.insert(AUTHORIZATION, b.clone());
            }
            h
        };
        let sent = oauth.bearer().await;
        let response = build(with_bearer(sent.as_ref()))
            .send()
            .await
            .with_context(&what)?;
        match response.status() {
            StatusCode::UNAUTHORIZED => {},
            StatusCode::FORBIDDEN => {
                if let Some(scope) = super::oauth::insufficient_scope(response.headers()) {
                    let scope = Some(scope).filter(|s| !s.is_empty());
                    if let Some(s) = &scope {
                        oauth.note_insufficient_scope(s).await;
                    }
                    return Err(oauth.sign_in_required(scope));
                }
                return Ok(response);
            },
            _ => return Ok(response),
        }
        let Some(fresh) = oauth.after_unauthorized(sent.as_ref()).await else {
            return Err(oauth.sign_in_required(None));
        };
        let retry = build(with_bearer(Some(&fresh)))
            .send()
            .await
            .with_context(&what)?;
        if retry.status() == StatusCode::UNAUTHORIZED {
            return Err(oauth.sign_in_required(None));
        }
        Ok(retry)
    }

    /// The header set for one request: static config headers, env-resolved
    /// headers, then session id / protocol version / Accept. Env values are
    /// re-read per request; a missing var is skipped, an un-headerable value
    /// errors naming only the header and the var.
    fn request_headers(&self, accept: Option<&'static str>) -> Result<HeaderMap> {
        let mut headers = self.static_headers.clone();
        for (name, var) in &self.env_headers {
            if let Ok(value) = std::env::var(var) {
                let v = HeaderValue::from_str(&value).map_err(|_| {
                    anyhow!("invalid value in env var '{var}' for MCP header '{name}'")
                })?;
                headers.insert(name.clone(), v);
            }
        }
        if let Some(accept) = accept {
            headers.insert(ACCEPT, HeaderValue::from_static(accept));
        }
        if let Some(sid) = self
            .session_id
            .read()
            .expect("mcp session_id lock poisoned")
            .clone()
        {
            headers.insert(SESSION_HEADER, sid);
        }
        if let Some(pv) = self
            .protocol_version
            .read()
            .expect("mcp protocol_version lock poisoned")
            .clone()
        {
            headers.insert(PROTOCOL_VERSION_HEADER, pv);
        }
        Ok(headers)
    }

    /// Capture `MCP-Session-Id` from any response's headers (the normative
    /// source is the `InitializeResult` response; accepting it from any response
    /// is a safe superset). Lookup is case-insensitive by `HeaderMap` contract.
    fn capture_session(&self, headers: &HeaderMap) {
        // 2026-07-28 has no sessions: a stray header is ignored.
        if self.is_modern() {
            return;
        }
        if let Some(v) = headers.get(&SESSION_HEADER) {
            *self
                .session_id
                .write()
                .expect("mcp session_id lock poisoned") = Some(v.clone());
        }
    }

    /// Build the error for a non-success status: the 404-with-session case is
    /// session expiry (spec says the client MUST re-initialize; we surface it
    /// instead of silently re-initializing mid-run — server-side tool state is
    /// gone, so the user should restart the connection deliberately). Other
    /// statuses carry a short redacted body snippet.
    async fn status_error(
        &self,
        method: &str,
        status: StatusCode,
        response: reqwest::Response,
    ) -> anyhow::Error {
        if status == StatusCode::NOT_FOUND
            && self
                .session_id
                .read()
                .expect("mcp session_id lock poisoned")
                .is_some()
        {
            return anyhow!(
                "MCP session expired (HTTP 404); a new session must be initialized — \
                 restart the server connection (method: {method})"
            );
        }
        let body = response.text().await.unwrap_or_default();
        let snippet = mermaid_model::utils::redact_secrets(&body);
        let end = snippet.floor_char_boundary(ERROR_BODY_SNIPPET_BYTES.min(snippet.len()));
        // A 2026-07-28 server puts a JSON-RPC error in a 4xx body (version,
        // header and capability errors); keep it typed for era detection.
        let rpc = serde_json::from_str::<Value>(&body)
            .ok()
            .and_then(|v| v.get("error").map(JsonRpcError::from_error_member));
        HttpStatusError {
            status: status.as_u16(),
            rpc,
            text: format!(
                "MCP server returned HTTP {} (method: {}): {}",
                status,
                method,
                &snippet[..end]
            ),
        }
        .into()
    }
}

/// A non-success HTTP status from the MCP endpoint, with the JSON-RPC error
/// its body carried, if any.
#[derive(Debug)]
pub(super) struct HttpStatusError {
    pub status: u16,
    pub rpc: Option<JsonRpcError>,
    text: String,
}

impl std::fmt::Display for HttpStatusError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.text)
    }
}

impl std::error::Error for HttpStatusError {}

/// Tracks the SSE `id:` and `retry:` fields, which [`drain_sse_events`]
/// deliberately discards. Fed the same raw bytes as the event buffer; only
/// complete lines are inspected, so a field split across chunks still parses.
#[derive(Default)]
struct SseMeta {
    line_buf: Vec<u8>,
    /// `id:` of the event currently being parsed. Promoted to
    /// `last_event_id` only at the event's terminating blank line — WHATWG
    /// SSE commits the cursor at dispatch, not at field parse, so a
    /// disconnect mid-event must not advance the resume point past the very
    /// event that was cut off.
    pending_event_id: Option<String>,
    /// Last fully-delivered event id — the SEP-1699 resumption cursor.
    last_event_id: Option<String>,
    /// Server-requested reconnection delay in milliseconds.
    retry_ms: Option<u64>,
}

impl SseMeta {
    /// Reset per-stream parse state before draining a (re)connected stream.
    /// The resumption cursor and retry delay survive reconnects; a partial
    /// line cut off by the disconnect must not prefix the resumed stream's
    /// first line and corrupt id/retry parsing.
    fn start_stream(&mut self) {
        self.line_buf.clear();
        self.pending_event_id = None;
    }

    fn feed(&mut self, chunk: &[u8]) {
        self.line_buf.extend_from_slice(chunk);
        while let Some(pos) = self.line_buf.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = self.line_buf.drain(..=pos).collect();
            let text = String::from_utf8_lossy(&line);
            let text = text.trim_end_matches(['\n', '\r']);
            if text.is_empty() {
                // Event boundary: the event was fully delivered.
                if let Some(id) = self.pending_event_id.take() {
                    self.last_event_id = Some(id);
                }
            } else if let Some(v) = text.strip_prefix("id:") {
                // Per SSE spec, one leading space after the colon is ignored,
                // and an empty id (the SEP-1699 priming event) still counts as
                // "this stream has IDs" — record it.
                let v = v.strip_prefix(' ').unwrap_or(v);
                self.pending_event_id = Some(v.to_string());
            } else if let Some(v) = text.strip_prefix("retry:")
                && let Ok(ms) = v.trim().parse::<u64>()
            {
                self.retry_ms = Some(ms);
            }
        }
    }
}

/// Refuse an IP-literal host the private-network policy blocks. reqwest
/// connects to IP-literal hosts directly, never consulting the dns_resolver
/// of [`vetted_client`] — so literals are vetted here, and the policy also
/// holds for `url = "https://192.168.1.5/mcp"`. DNS names classify as Public
/// and are vetted at connect time by [`McpVettingResolver`].
pub(super) fn check_ip_literal(url: &reqwest::Url, allow_private: bool) -> Result<()> {
    let host = url.host_str().unwrap_or_default();
    let blocked = match classify_host(host) {
        HostClass::Loopback | HostClass::Public => false,
        _ => !allow_private,
    };
    if blocked {
        bail!(
            "refusing to connect MCP server '{host}': it is a private/internal \
             address (set allow_private_network = true for this server to permit it)"
        );
    }
    Ok(())
}

/// The HTTP client for one MCP server's traffic: its endpoint and, for
/// OAuth, the metadata, registration and token endpoints it names. Same
/// private-network policy for all of them, since the server chooses the
/// OAuth URLs.
pub(super) fn vetted_client(allow_private: bool) -> Result<reqwest::Client> {
    reqwest::Client::builder()
        .user_agent(format!("mermaid/{}", env!("CARGO_PKG_VERSION")))
        .connect_timeout(Duration::from_secs(CONNECT_TIMEOUT_SECS))
        // Following a redirect would forward Authorization headers to a
        // possibly different origin, and could reach an IP literal the
        // resolver never sees.
        .redirect(reqwest::redirect::Policy::none())
        .dns_resolver(std::sync::Arc::new(McpVettingResolver { allow_private }))
        .build()
        .context("failed to build MCP HTTP client")
}

/// Connect-time DNS vetting for MCP endpoints. Unlike `web_fetch`'s resolver,
/// loopback is always allowed (local MCP servers are a first-class case);
/// private / link-local / CGNAT / metadata addresses are blocked unless the
/// config opts in with `allow_private_network` — plugin bundles ship MCP
/// configs, so a malicious bundle must not reach 169.254.169.254 or the LAN.
struct McpVettingResolver {
    allow_private: bool,
}

impl reqwest::dns::Resolve for McpVettingResolver {
    fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
        let allow_private = self.allow_private;
        Box::pin(async move {
            let host = name.as_str().to_string();
            // Port 0 is a placeholder — reqwest overrides it with the URL's
            // port. We only need the resolved IPs in order to vet them.
            let addrs: Vec<std::net::SocketAddr> =
                tokio::net::lookup_host((host.as_str(), 0)).await?.collect();
            for addr in &addrs {
                let class = classify_host(&addr.ip().to_string());
                let blocked = match class {
                    HostClass::Loopback | HostClass::Public => false,
                    _ => !allow_private,
                };
                if blocked {
                    return Err(format!(
                        "refusing to connect MCP server '{host}': it resolves to a \
                         private/internal address (set allow_private_network = true \
                         for this server to permit it)"
                    )
                    .into());
                }
            }
            Ok(Box::new(addrs.into_iter()) as reqwest::dns::Addrs)
        })
    }
}

/// Canned-HTTP test fixture (project precedent: ollama/daemon tests), shared
/// with the `server_manager` tests.
///
/// A tokio `TcpListener` that serves one canned reply per accepted connection,
/// in order, recording each raw request. Every reply carries
/// `Connection: close`, so reqwest opens a fresh connection per request and
/// the accept loop stays strictly sequential.
#[cfg(test)]
pub(super) mod test_fixture {
    use super::*;
    use std::sync::Arc;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::sync::Mutex;

    pub enum Reply {
        Raw(String),
        /// Accept and read the request, then never respond.
        Hang,
    }

    pub struct Fixture {
        pub url: String,
        requests: Arc<Mutex<Vec<String>>>,
    }

    impl Fixture {
        pub async fn requests(&self) -> Vec<String> {
            self.requests.lock().await.clone()
        }

        pub fn config(&self) -> McpServerConfig {
            McpServerConfig {
                url: Some(self.url.clone()),
                ..Default::default()
            }
        }

        /// A transport whose OAuth token store is an empty in-memory one,
        /// so no test reads the real keyring.
        pub fn transport(&self) -> HttpTransport {
            HttpTransport::with_store(
                "fx",
                &self.config(),
                Some(Arc::new(super::super::oauth::MemStore::default())),
            )
            .expect("transport")
        }
    }

    pub async fn fixture(replies: Vec<Reply>) -> Fixture {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("addr");
        let requests = Arc::new(Mutex::new(Vec::new()));
        let recorded = Arc::clone(&requests);
        tokio::spawn(async move {
            for reply in replies {
                let Ok((mut sock, _)) = listener.accept().await else {
                    return;
                };
                let raw = read_http_request(&mut sock).await;
                recorded.lock().await.push(raw);
                match reply {
                    Reply::Raw(bytes) => {
                        let _ = sock.write_all(bytes.as_bytes()).await;
                        let _ = sock.shutdown().await;
                    },
                    Reply::Hang => {
                        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                    },
                }
            }
        });
        Fixture {
            url: format!("http://127.0.0.1:{}/mcp", addr.port()),
            requests,
        }
    }

    /// Read one HTTP/1.1 request: headers through `\r\n\r\n`, then a body of
    /// exactly `Content-Length` bytes.
    async fn read_http_request(sock: &mut tokio::net::TcpStream) -> String {
        let mut buf: Vec<u8> = Vec::new();
        let mut tmp = [0u8; 4096];
        loop {
            if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                let head = String::from_utf8_lossy(&buf[..pos]).to_ascii_lowercase();
                let content_length = head
                    .lines()
                    .find_map(|l| l.strip_prefix("content-length:"))
                    .and_then(|v| v.trim().parse::<usize>().ok())
                    .unwrap_or(0);
                let total = pos + 4 + content_length;
                while buf.len() < total {
                    match sock.read(&mut tmp).await {
                        Ok(0) | Err(_) => break,
                        Ok(n) => buf.extend_from_slice(&tmp[..n]),
                    }
                }
                return String::from_utf8_lossy(&buf[..total.min(buf.len())]).into_owned();
            }
            match sock.read(&mut tmp).await {
                Ok(0) | Err(_) => return String::from_utf8_lossy(&buf).into_owned(),
                Ok(n) => buf.extend_from_slice(&tmp[..n]),
            }
        }
    }

    pub fn json_reply_with_headers(body: &str, extra_headers: &str) -> Reply {
        Reply::Raw(format!(
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n{extra_headers}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        ))
    }

    pub fn json_reply(body: &str) -> Reply {
        json_reply_with_headers(body, "")
    }

    pub fn sse_reply(events: &str) -> Reply {
        Reply::Raw(format!(
            "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{events}",
            events.len()
        ))
    }

    pub fn status_reply(code: u16, reason: &str) -> Reply {
        Reply::Raw(format!(
            "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
        ))
    }

    pub fn rpc_response(id: u64, result: &str) -> String {
        format!(r#"{{"jsonrpc":"2.0","id":{id},"result":{result}}}"#)
    }
}

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

    #[tokio::test]
    async fn plain_json_response_round_trips() {
        let fx = fixture(vec![json_reply(&rpc_response(1, r#"{"ok":true}"#))]).await;
        let t = fx.transport();
        let result = t.send_request("ping", json!({})).await.expect("result");
        assert_eq!(result["ok"], true);
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 1);
        // POST with the Accept pair, JSON body carrying our method.
        assert!(reqs[0].starts_with("POST /mcp"), "{}", reqs[0]);
        assert!(
            reqs[0]
                .to_ascii_lowercase()
                .contains("accept: application/json, text/event-stream"),
            "{}",
            reqs[0]
        );
        assert!(reqs[0].contains(r#""method":"ping""#), "{}", reqs[0]);
    }

    #[tokio::test]
    async fn sse_response_drains_notifications_until_our_id() {
        let events = format!(
            "data: {}\n\ndata: {}\n\n",
            r#"{"jsonrpc":"2.0","method":"notifications/message","params":{}}"#,
            rpc_response(1, r#"{"via":"sse"}"#)
        );
        let fx = fixture(vec![sse_reply(&events)]).await;
        let t = fx.transport();
        let result = t
            .send_request("tools/list", json!({}))
            .await
            .expect("result");
        assert_eq!(result["via"], "sse");
    }

    #[tokio::test]
    async fn session_id_is_captured_and_echoed_including_delete() {
        let fx = fixture(vec![
            // Old-spelling header on purpose: capture must be case-insensitive.
            json_reply_with_headers(&rpc_response(1, "{}"), "Mcp-Session-Id: sess-123\r\n"),
            json_reply(&rpc_response(2, "{}")),
            status_reply(200, "OK"), // DELETE
        ])
        .await;
        let t = fx.transport();
        t.send_request("initialize", json!({})).await.expect("init");
        t.send_request("tools/list", json!({})).await.expect("list");
        t.shutdown().await;
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 3);
        assert!(
            !reqs[0].to_ascii_lowercase().contains("mcp-session-id"),
            "no session id before the server assigns one: {}",
            reqs[0]
        );
        assert!(
            reqs[1]
                .to_ascii_lowercase()
                .contains("mcp-session-id: sess-123"),
            "{}",
            reqs[1]
        );
        assert!(reqs[2].starts_with("DELETE /mcp"), "{}", reqs[2]);
        assert!(
            reqs[2]
                .to_ascii_lowercase()
                .contains("mcp-session-id: sess-123"),
            "{}",
            reqs[2]
        );
    }

    #[tokio::test]
    async fn shutdown_without_session_sends_nothing() {
        let fx = fixture(vec![]).await;
        let t = fx.transport();
        t.shutdown().await;
        assert!(fx.requests().await.is_empty());
    }

    #[tokio::test]
    async fn delete_405_is_tolerated() {
        let fx = fixture(vec![
            json_reply_with_headers(&rpc_response(1, "{}"), "MCP-Session-Id: s1\r\n"),
            status_reply(405, "Method Not Allowed"),
        ])
        .await;
        let t = fx.transport();
        t.send_request("initialize", json!({})).await.expect("init");
        // Must not panic or error — shutdown is infallible by design.
        t.shutdown().await;
        assert_eq!(fx.requests().await.len(), 2);
    }

    #[tokio::test]
    async fn http_404_with_active_session_reports_expiry() {
        let fx = fixture(vec![
            json_reply_with_headers(&rpc_response(1, "{}"), "MCP-Session-Id: s1\r\n"),
            status_reply(404, "Not Found"),
        ])
        .await;
        let t = fx.transport();
        t.send_request("initialize", json!({})).await.expect("init");
        let err = t
            .send_request("tools/list", json!({}))
            .await
            .expect_err("expired");
        assert!(err.to_string().contains("session expired"), "{err}");
    }

    #[tokio::test]
    async fn static_and_env_headers_reach_the_wire() {
        // Unique env var name — nextest runs tests in parallel processes but
        // in-process parallelism still forbids reuse across tests.
        const VAR: &str = "MERMAID_TEST_MCP_HTTP_ENV_HEADER_A";
        unsafe { std::env::set_var(VAR, "from-env") };
        let fx = fixture(vec![json_reply(&rpc_response(1, "{}"))]).await;
        let mut config = fx.config();
        config
            .headers
            .insert("X-Static-Token".to_string(), "static-secret".to_string());
        config
            .env_headers
            .insert("X-Env-Token".to_string(), VAR.to_string());
        // A missing env var is skipped, not an error.
        config.env_headers.insert(
            "X-Missing".to_string(),
            "MERMAID_TEST_MCP_HTTP_ENV_HEADER_MISSING".to_string(),
        );
        let t = HttpTransport::with_store(
            "t",
            &config,
            Some(std::sync::Arc::new(super::super::oauth::MemStore::default())),
        )
        .expect("transport");
        t.send_request("ping", json!({})).await.expect("result");
        let req = fx.requests().await.remove(0).to_ascii_lowercase();
        assert!(req.contains("x-static-token: static-secret"), "{req}");
        assert!(req.contains("x-env-token: from-env"), "{req}");
        assert!(!req.contains("x-missing"), "{req}");
    }

    #[tokio::test]
    async fn invalid_header_name_errors_at_construction_without_value() {
        let mut config = McpServerConfig {
            url: Some("https://example.com/mcp".to_string()),
            ..Default::default()
        };
        config
            .headers
            .insert("bad header".to_string(), "secret-value".to_string());
        // map to () — HttpTransport has no Debug (it holds secret headers).
        let err = HttpTransport::new("t", &config)
            .map(|_| ())
            .expect_err("bad name");
        assert!(err.to_string().contains("bad header"), "{err}");
        assert!(!err.to_string().contains("secret-value"), "{err}");
    }

    #[tokio::test]
    async fn notification_accepts_202_and_rejects_500() {
        let fx = fixture(vec![
            status_reply(202, "Accepted"),
            Reply::Raw(
                "HTTP/1.1 500 Internal Server Error\r\nContent-Type: text/plain\r\nContent-Length: 4\r\nConnection: close\r\n\r\noops"
                    .to_string(),
            ),
        ])
        .await;
        let t = fx.transport();
        t.send_notification("notifications/initialized", json!({}))
            .await
            .expect("202 ok");
        let err = t
            .send_notification("notifications/initialized", json!({}))
            .await
            .expect_err("500 err");
        // The status lives in the context chain; `:#` renders the whole chain.
        let rendered = format!("{err:#}");
        assert!(rendered.contains("500"), "{rendered}");
        assert!(
            rendered.contains("oops"),
            "body snippet expected: {rendered}"
        );
    }

    #[tokio::test]
    async fn http_202_on_a_request_is_a_protocol_error() {
        let fx = fixture(vec![status_reply(202, "Accepted")]).await;
        let t = fx.transport();
        let err = t
            .send_request("tools/list", json!({}))
            .await
            .expect_err("202 on request");
        assert!(err.to_string().contains("202"), "{err}");
    }

    #[tokio::test]
    async fn sse_eof_without_response_and_without_ids_errors() {
        let events =
            "data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\",\"params\":{}}\n\n";
        let fx = fixture(vec![sse_reply(events)]).await;
        let t = fx.transport();
        let err = t
            .send_request("tools/list", json!({}))
            .await
            .expect_err("eof");
        assert!(
            err.to_string().contains("ended without a response"),
            "{err}"
        );
    }

    #[tokio::test]
    async fn sse_disconnect_with_event_id_resumes_via_get_last_event_id() {
        // SEP-1699 polling: stream 1 sends a priming event (id, empty data)
        // plus a retry hint, then closes without the response. The client must
        // wait and resume with GET + Last-Event-ID; stream 2 delivers the
        // response.
        let priming = "id: ev-7\nretry: 10\ndata:\n\n";
        let resumed = format!(
            "id: ev-8\ndata: {}\n\n",
            rpc_response(1, r#"{"resumed":true}"#)
        );
        let fx = fixture(vec![sse_reply(priming), sse_reply(&resumed)]).await;
        let t = fx.transport();
        let result = t
            .send_request("tools/call", json!({}))
            .await
            .expect("resumed");
        assert_eq!(result["resumed"], true);
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 2);
        assert!(reqs[1].starts_with("GET /mcp"), "{}", reqs[1]);
        let get = reqs[1].to_ascii_lowercase();
        assert!(get.contains("last-event-id: ev-7"), "{}", reqs[1]);
        assert!(get.contains("accept: text/event-stream"), "{}", reqs[1]);
    }

    #[tokio::test]
    async fn server_request_on_sse_stream_gets_error_reply_posted_back() {
        let events = format!(
            "data: {}\n\ndata: {}\n\n",
            r#"{"jsonrpc":"2.0","id":99,"method":"sampling/createMessage","params":{}}"#,
            rpc_response(1, "{}")
        );
        let fx = fixture(vec![sse_reply(&events), status_reply(202, "Accepted")]).await;
        let t = fx.transport();
        t.send_request("tools/list", json!({}))
            .await
            .expect("result");
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 2, "the -32601 reply must be POSTed back");
        assert!(reqs[1].contains("-32601"), "{}", reqs[1]);
        assert!(reqs[1].contains(r#""id":99"#), "{}", reqs[1]);
    }

    #[tokio::test]
    async fn server_ping_on_sse_stream_gets_result_not_32601() {
        let events = format!(
            "data: {}\n\ndata: {}\n\n",
            r#"{"jsonrpc":"2.0","id":42,"method":"ping"}"#,
            rpc_response(1, "{}")
        );
        let fx = fixture(vec![sse_reply(&events), status_reply(202, "Accepted")]).await;
        let t = fx.transport();
        t.send_request("tools/list", json!({}))
            .await
            .expect("result");
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 2);
        assert!(reqs[1].contains(r#""result":{}"#), "{}", reqs[1]);
        assert!(!reqs[1].contains("-32601"), "{}", reqs[1]);
    }

    #[tokio::test]
    async fn injected_short_timeout_is_enforced() {
        let fx = fixture(vec![Reply::Hang]).await;
        let t = fx.transport();
        let start = std::time::Instant::now();
        let err = t
            .send_request_with_timeout("tools/list", json!({}), 1)
            .await
            .expect_err("must time out");
        assert!(err.to_string().contains("timed out"), "{err}");
        assert!(start.elapsed() < std::time::Duration::from_secs(5));
    }

    /// The 400 a 2025-11-25 TypeScript-SDK server gives a request without a
    /// session: a JSON-RPC error, but not a 2026-07-28 one.
    fn legacy_no_session() -> Reply {
        Reply::Raw({
            let body = r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32000,"message":"Bad Request: No valid session ID provided"}}"#;
            format!(
                "HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            )
        })
    }

    fn rpc_error_reply(status: u16, code: i64, data: &str) -> Reply {
        let body = format!(
            r#"{{"jsonrpc":"2.0","id":1,"error":{{"code":{code},"message":"m","data":{data}}}}}"#
        );
        Reply::Raw(format!(
            "HTTP/1.1 {status} Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        ))
    }

    const DISCOVER_MODERN: &str = r#"{"resultType":"complete","supportedVersions":["2026-07-28"],"capabilities":{"tools":{}},"_meta":{"io.modelcontextprotocol/serverInfo":{"name":"modern-fx","version":"2.0"}}}"#;

    #[tokio::test]
    async fn legacy_server_falls_back_to_initialize_with_session_headers() {
        let init_result = r#"{"protocolVersion":"2025-11-25","capabilities":{},"serverInfo":{"name":"fx","version":"1.0"}}"#;
        let fx = fixture(vec![
            legacy_no_session(),
            json_reply_with_headers(&rpc_response(2, init_result), "MCP-Session-Id: s9\r\n"),
            status_reply(202, "Accepted"), // notifications/initialized
            json_reply(&rpc_response(3, r#"{"tools":[]}"#)),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        let info = client.initialize().await.expect("initialize");
        assert_eq!(info.name, "fx");
        assert_eq!(client.era(), super::super::client::Era::Legacy);
        client.list_tools().await.expect("list");
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 4);
        // The probe is a 2026-07-28 request.
        let probe = reqs[0].to_ascii_lowercase();
        assert!(
            probe.contains("mcp-protocol-version: 2026-07-28"),
            "{probe}"
        );
        assert!(probe.contains("mcp-method: server/discover"), "{probe}");
        assert!(
            reqs[0].contains(r#""io.modelcontextprotocol/protocolVersion":"2026-07-28""#),
            "{}",
            reqs[0]
        );
        // initialize itself carries no version header.
        let init = reqs[1].to_ascii_lowercase();
        assert!(!init.contains("mcp-protocol-version"), "{init}");
        assert!(!init.contains("mcp-method"), "{init}");
        // Everything after initialize — including notifications/initialized —
        // carries the negotiated version and the session id, and no _meta.
        for req in &reqs[2..] {
            let low = req.to_ascii_lowercase();
            assert!(low.contains("mcp-protocol-version: 2025-11-25"), "{req}");
            assert!(low.contains("mcp-session-id: s9"), "{req}");
            assert!(!req.contains("io.modelcontextprotocol/"), "{req}");
        }
    }

    #[tokio::test]
    async fn modern_server_skips_the_handshake_and_mirrors_headers() {
        let tools = r#"{"resultType":"complete","tools":[
            {"name":"execute_sql","inputSchema":{"type":"object","properties":{
                "region":{"type":"string","x-mcp-header":"Region"},
                "query":{"type":"string"}}}},
            {"name":"broken","inputSchema":{"type":"object","properties":{
                "n":{"type":"number","x-mcp-header":"N"}}}}
        ]}"#;
        let call = r#"{"resultType":"complete","content":[{"type":"text","text":"ok"}]}"#;
        let fx = fixture(vec![
            // A stray session header must be ignored in 2026-07-28.
            json_reply_with_headers(&rpc_response(1, DISCOVER_MODERN), "MCP-Session-Id: s9\r\n"),
            json_reply(&rpc_response(2, tools)),
            json_reply(&rpc_response(3, call)),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        let info = client.initialize().await.expect("initialize");
        assert_eq!(info.name, "modern-fx");
        assert_eq!(client.era(), super::super::client::Era::Modern);
        let listed = client.list_tools().await.expect("list");
        let names: Vec<&str> = listed.iter().map(|t| t.name.as_str()).collect();
        assert_eq!(names, ["execute_sql"], "the invalid tool is left out");
        let result = client
            .call_tool(
                "execute_sql",
                &json!({"region": "us-west1", "query": "SELECT 1"}),
            )
            .await
            .expect("call");
        assert!(!result.is_error);

        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 3, "no initialize, no notification");
        for req in &reqs {
            let low = req.to_ascii_lowercase();
            assert!(low.contains("mcp-protocol-version: 2026-07-28"), "{req}");
            assert!(!low.contains("mcp-session-id"), "{req}");
            assert!(
                req.contains(r#""io.modelcontextprotocol/protocolVersion":"2026-07-28""#),
                "{req}"
            );
            assert!(
                req.contains(r#""io.modelcontextprotocol/clientCapabilities":{}"#),
                "{req}"
            );
        }
        let call = reqs[2].to_ascii_lowercase();
        assert!(call.contains("mcp-method: tools/call"), "{call}");
        assert!(call.contains("mcp-name: execute_sql"), "{call}");
        assert!(call.contains("mcp-param-region: us-west1"), "{call}");
    }

    #[tokio::test]
    async fn unsupported_version_with_a_legacy_entry_uses_the_handshake() {
        let init_result =
            r#"{"protocolVersion":"2025-11-25","capabilities":{},"serverInfo":{"name":"fx"}}"#;
        let fx = fixture(vec![
            rpc_error_reply(
                400,
                -32022,
                r#"{"supported":["2025-11-25"],"requested":"2026-07-28"}"#,
            ),
            json_reply(&rpc_response(2, init_result)),
            status_reply(202, "Accepted"),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        client.initialize().await.expect("initialize");
        assert_eq!(client.era(), super::super::client::Era::Legacy);
    }

    #[tokio::test]
    async fn unsupported_version_with_no_common_version_is_an_error() {
        let fx = fixture(vec![rpc_error_reply(
            400,
            -32022,
            r#"{"supported":["2027-03-01"],"requested":"2026-07-28"}"#,
        )])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        let err = client.initialize().await.expect_err("no common version");
        assert!(err.to_string().contains("2027-03-01"), "{err}");
        assert_eq!(fx.requests().await.len(), 1, "no fallback to initialize");
    }

    #[tokio::test]
    async fn sign_in_needed_on_the_probe_is_not_a_legacy_server() {
        let fx = fixture(vec![Reply::Raw(
            "HTTP/1.1 401 Unauthorized\r\nWWW-Authenticate: Bearer\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
                .to_string(),
        )])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        let err = client.initialize().await.expect_err("sign-in needed");
        assert!(super::super::oauth::is_auth_required(&err), "{err:#}");
        assert_eq!(fx.requests().await.len(), 1, "no fallback to initialize");
    }

    #[tokio::test]
    async fn input_required_with_state_only_is_retried_with_that_state() {
        let fx = fixture(vec![
            json_reply(&rpc_response(1, DISCOVER_MODERN)),
            json_reply(&rpc_response(
                2,
                r#"{"resultType":"input_required","requestState":"opaque-1"}"#,
            )),
            json_reply(&rpc_response(
                3,
                r#"{"resultType":"complete","content":[{"type":"text","text":"done"}]}"#,
            )),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        client.initialize().await.expect("initialize");
        let result = client.call_tool("t", &json!({})).await.expect("call");
        assert!(
            matches!(&result.content[0], super::super::client::ContentBlock::Text(t) if t == "done")
        );
        let reqs = fx.requests().await;
        assert!(!reqs[1].contains("requestState"), "{}", reqs[1]);
        assert!(
            reqs[2].contains(r#""requestState":"opaque-1""#),
            "{}",
            reqs[2]
        );
    }

    #[tokio::test]
    async fn input_required_that_asks_questions_is_an_error() {
        let fx = fixture(vec![
            json_reply(&rpc_response(1, DISCOVER_MODERN)),
            json_reply(&rpc_response(
                2,
                r#"{"resultType":"input_required","inputRequests":{"q":{"method":"elicitation/create","params":{}}}}"#,
            )),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        client.initialize().await.expect("initialize");
        let err = client.call_tool("t", &json!({})).await.expect_err("asks");
        assert!(err.to_string().contains("elicitation/create"), "{err}");
    }

    #[tokio::test]
    async fn header_mismatch_reloads_the_tool_list_and_retries() {
        let old_tools = r#"{"tools":[{"name":"t","inputSchema":{"type":"object"}}]}"#;
        let new_tools = r#"{"tools":[{"name":"t","inputSchema":{"type":"object","properties":{
            "id":{"type":"string","x-mcp-header":"Id"}}}}]}"#;
        let fx = fixture(vec![
            json_reply(&rpc_response(1, DISCOVER_MODERN)),
            json_reply(&rpc_response(2, old_tools)),
            rpc_error_reply(400, -32020, "null"),
            json_reply(&rpc_response(4, new_tools)),
            json_reply(&rpc_response(5, r#"{"content":[]}"#)),
        ])
        .await;
        let mut client = McpClient::new(fx.transport().into());
        client.initialize().await.expect("initialize");
        client.list_tools().await.expect("list");
        client
            .call_tool("t", &json!({"id": "abc"}))
            .await
            .expect("retried call");
        let reqs = fx.requests().await;
        assert_eq!(reqs.len(), 5);
        assert!(!reqs[2].to_ascii_lowercase().contains("mcp-param-id"));
        assert!(reqs[4].to_ascii_lowercase().contains("mcp-param-id: abc"));
    }

    #[tokio::test]
    async fn resolver_allows_loopback_blocks_private_honors_flag() {
        use reqwest::dns::Resolve;
        use std::str::FromStr;
        let name = |s: &str| reqwest::dns::Name::from_str(s).expect("name");
        // Loopback is always allowed — local MCP servers are first-class.
        let default = McpVettingResolver {
            allow_private: false,
        };
        assert!(default.resolve(name("localhost")).await.is_ok());
        // A private address is blocked by default...
        assert!(default.resolve(name("192.168.1.5")).await.is_err());
        // ...and allowed with the explicit opt-in.
        let opted_in = McpVettingResolver {
            allow_private: true,
        };
        assert!(opted_in.resolve(name("192.168.1.5")).await.is_ok());
    }

    #[test]
    fn sse_meta_tracks_id_and_retry_across_chunk_splits() {
        let mut meta = SseMeta::default();
        // Field split across chunks: only complete lines are inspected.
        meta.feed(b"id: ev");
        assert_eq!(meta.last_event_id, None);
        meta.feed(b"-1\nretry: 250\ndata: x\n\n");
        assert_eq!(meta.last_event_id.as_deref(), Some("ev-1"));
        assert_eq!(meta.retry_ms, Some(250));
        // A later id supersedes; CRLF line endings parse too.
        meta.feed(b"id: ev-2\r\ndata: y\r\n\r\n");
        assert_eq!(meta.last_event_id.as_deref(), Some("ev-2"));
    }

    #[test]
    fn sse_meta_commits_id_only_at_event_boundary() {
        let mut meta = SseMeta::default();
        meta.feed(b"id: ev-1\ndata: x\n\n");
        assert_eq!(meta.last_event_id.as_deref(), Some("ev-1"));
        // An event cut off before its terminating blank line must NOT advance
        // the cursor — resuming after it would skip its redelivery.
        meta.feed(b"id: ev-2\ndata: {\"jsonr");
        assert_eq!(meta.last_event_id.as_deref(), Some("ev-1"));
        // The resumed stream starts fresh: the partial line is discarded and
        // must not swallow the first field of the new stream.
        meta.start_stream();
        meta.feed(b"id: ev-2\ndata: y\n\n");
        assert_eq!(meta.last_event_id.as_deref(), Some("ev-2"));
    }

    #[test]
    fn new_vets_ip_literal_hosts() {
        // reqwest never consults the dns_resolver for IP literals, so new()
        // must apply the private-network policy itself.
        let private = McpServerConfig {
            url: Some("https://192.168.1.5/mcp".to_string()),
            ..Default::default()
        };
        // No expect_err: HttpTransport deliberately has no Debug impl (it
        // holds secret-bearing headers).
        let err = match HttpTransport::new("t", &private) {
            Ok(_) => panic!("private literal must be rejected"),
            Err(e) => e,
        };
        assert!(err.to_string().contains("allow_private_network"), "{err}");
        let metadata = McpServerConfig {
            url: Some("https://[::ffff:169.254.169.254]/mcp".to_string()),
            ..Default::default()
        };
        assert!(HttpTransport::new("t", &metadata).is_err());
        let opted_in = McpServerConfig {
            url: Some("https://192.168.1.5/mcp".to_string()),
            allow_private_network: true,
            ..Default::default()
        };
        assert!(HttpTransport::new("t", &opted_in).is_ok());
        // Loopback literals stay first-class.
        let loopback = McpServerConfig {
            url: Some("http://127.0.0.1:9099/mcp".to_string()),
            ..Default::default()
        };
        assert!(HttpTransport::new("t", &loopback).is_ok());
    }

    #[test]
    fn new_rejects_config_without_url() {
        let config = McpServerConfig {
            command: "npx".to_string(),
            ..Default::default()
        };
        assert!(HttpTransport::new("t", &config).is_err());
    }
}