use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;

use axum::Json;
use axum::body::Bytes;
use axum::extract::{Form, Query, State};
use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Router, extract::Request};
use dashmap::DashMap;
use rmcp::ErrorData as McpError;
use rmcp::model::*;
use serde_json::json;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;
use tower_http::limit::RequestBodyLimitLayer;

use super::error::HttpError;
use super::sse::sse_stream_with_heartbeat;
use crate::downstream_oauth::resource_scopes;
use crate::mcp_http_headers::{HEADER_MISMATCH_CODE, HeaderMismatch, validate_mirrored_headers};
use crate::notifications::{NotificationTarget, ProtocolNotification};
use crate::proxy::{DownstreamBridge, DownstreamCallContext, ToolRouter};
use crate::session::{SessionSendOutcome, SessionStore, SseMessage, SseReplayKey};

/// rmcp header constant for session ID.
const SESSION_ID_HEADER: &str = "Mcp-Session-Id";

/// MCP protocol version header name.
const PROTOCOL_VERSION_HEADER: &str = "MCP-Protocol-Version";
const TRACEPARENT_HEADER: &str = "traceparent";
const PLUG_TRACE_ID_HEADER: &str = "x-plug-trace-id";

/// The MCP protocol version we implement.
const PROTOCOL_VERSION: &str = "2025-11-25";

/// Shared state for all HTTP handlers.
pub struct HttpState {
    pub router: Arc<ToolRouter>,
    pub sessions: Arc<dyn SessionStore>,
    pub cancel: CancellationToken,
    pub auth_mode: crate::config::DownstreamAuthMode,
    pub downstream_oauth: Option<crate::downstream_oauth::DownstreamOauthManager>,
    pub sse_channel_capacity: usize,
    pub allowed_origins: Vec<Arc<str>>,
    pub notification_task_started: AtomicBool,
    /// Bearer token for downstream client authentication.
    /// `None` means the current downstream auth mode does not require bearer auth.
    pub auth_token: Option<Arc<str>>,
    /// Sessions whose clients advertise `roots` capability.
    pub roots_capable_sessions: DashMap<String, ()>,
    /// Pending reverse requests sent to HTTP clients (keyed by session_id + request_id).
    pub pending_client_requests: DashMap<(String, i64), oneshot::Sender<ClientResult>>,
    /// Counter for generating unique reverse-request IDs.
    pub reverse_request_counter: AtomicU64,
    /// Per-session client capabilities for reverse-request gating.
    pub client_capabilities: DashMap<String, ClientCapabilities>,
}

/// HTTP-specific bridge for forwarding reverse requests (elicitation, sampling)
/// to a downstream HTTP client via its SSE stream.
struct HttpBridge {
    state: Arc<HttpState>,
    session_id: Arc<str>,
    capabilities: ClientCapabilities,
}

impl DownstreamBridge for HttpBridge {
    fn create_elicitation(
        &self,
        request: CreateElicitationRequestParams,
    ) -> Pin<Box<dyn Future<Output = Result<CreateElicitationResult, McpError>> + Send + '_>> {
        if self.capabilities.elicitation.is_none() {
            return Box::pin(async {
                Err(McpError::internal_error(
                    "client does not support elicitation".to_string(),
                    None,
                ))
            });
        }
        let state = Arc::clone(&self.state);
        let session_id = Arc::clone(&self.session_id);
        Box::pin(async move {
            let result = send_http_client_request(
                &state,
                &session_id,
                ServerRequest::CreateElicitationRequest(CreateElicitationRequest::new(request)),
                Some(Duration::from_secs(600)), // 10-minute upper bound prevents resource leaks
            )
            .await?;
            match result {
                ClientResult::CreateElicitationResult(r) => Ok(r),
                other => Err(McpError::internal_error(
                    format!("unexpected elicitation response: {other:?}"),
                    None,
                )),
            }
        })
    }

    fn create_message(
        &self,
        request: CreateMessageRequestParams,
    ) -> Pin<Box<dyn Future<Output = Result<CreateMessageResult, McpError>> + Send + '_>> {
        if self.capabilities.sampling.is_none() {
            return Box::pin(async {
                Err(McpError::internal_error(
                    "client does not support sampling".to_string(),
                    None,
                ))
            });
        }
        let state = Arc::clone(&self.state);
        let session_id = Arc::clone(&self.session_id);
        Box::pin(async move {
            let result = send_http_client_request(
                &state,
                &session_id,
                ServerRequest::CreateMessageRequest(CreateMessageRequest::new(request)),
                Some(Duration::from_secs(60)), // sampling has bounded timeout
            )
            .await?;
            match result {
                ClientResult::CreateMessageResult(r) => Ok(*r),
                other => Err(McpError::internal_error(
                    format!("unexpected sampling response: {other:?}"),
                    None,
                )),
            }
        })
    }
}

impl HttpState {
    pub fn spawn_notification_fanout(self: &Arc<Self>) {
        if self
            .notification_task_started
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            return;
        }

        let state = Arc::clone(self);
        let mut rx = state.router.subscribe_notifications();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    biased;
                    _ = state.cancel.cancelled() => break,
                    recv = rx.recv() => {
                        match recv {
                            Ok(notification) => {
                                match notification {
                                    ProtocolNotification::ToolListChanged => {
                                        if let Some(message) = notification_to_sse_message(
                                            ProtocolNotification::ToolListChanged,
                                        ) {
                                            state.sessions.broadcast(message);
                                        }
                                    }
                                    ProtocolNotification::ToolListChangedFor { target } => {
                                        if let NotificationTarget::Http { session_id } = target {
                                            let session_key = session_id.to_string();
                                            if let Some(message) = notification_to_sse_message(
                                                ProtocolNotification::ToolListChangedFor {
                                                    target: NotificationTarget::Http {
                                                        session_id,
                                                    },
                                                },
                                            ) {
                                                state.sessions.send_to_session(&session_key, message);
                                            }
                                        }
                                    }
                                    ProtocolNotification::ResourceListChanged => {
                                        if let Some(message) = notification_to_sse_message(
                                            ProtocolNotification::ResourceListChanged,
                                        ) {
                                            state.sessions.broadcast(message);
                                        }
                                    }
                                    ProtocolNotification::PromptListChanged => {
                                        if let Some(message) = notification_to_sse_message(
                                            ProtocolNotification::PromptListChanged,
                                        ) {
                                            state.sessions.broadcast(message);
                                        }
                                    }
                                    ProtocolNotification::Progress { target, params } => {
                                        if let NotificationTarget::Http { session_id } = target {
                                            let session_key = session_id.to_string();
                                            if let Some(message) = notification_to_sse_message(
                                                ProtocolNotification::Progress {
                                                    target: NotificationTarget::Http {
                                                        session_id,
                                                    },
                                                    params,
                                                },
                                            ) {
                                                state.sessions.send_to_session(&session_key, message);
                                            }
                                        }
                                    }
                                    ProtocolNotification::Cancelled { target, params } => {
                                        if let NotificationTarget::Http { session_id } = target {
                                            let session_key = session_id.to_string();
                                            if let Some(message) = notification_to_sse_message(
                                                ProtocolNotification::Cancelled {
                                                    target: NotificationTarget::Http {
                                                        session_id,
                                                    },
                                                    params,
                                                },
                                            ) {
                                                state.sessions.send_to_session(&session_key, message);
                                            }
                                        }
                                    }
                                    ProtocolNotification::ResourceUpdated { target, params } => {
                                        if let NotificationTarget::Http { session_id } = target {
                                            let session_key = session_id.to_string();
                                            if let Some(message) = notification_to_sse_message(
                                                ProtocolNotification::ResourceUpdated {
                                                    target: NotificationTarget::Http {
                                                        session_id,
                                                    },
                                                    params,
                                                },
                                            ) {
                                                state.sessions.send_to_session(&session_key, message);
                                            }
                                        }
                                    }
                                    ref notification @ (
                                        ProtocolNotification::LoggingMessage { .. }
                                        | ProtocolNotification::TokenRefreshExchanged { .. }
                                        | ProtocolNotification::AuthStateChanged { .. }
                                    ) => {
                                        if let Some(params) = notification.as_logging_message_params() {
                                            if let Some(message) = notification_to_sse_message(
                                                ProtocolNotification::LoggingMessage { params },
                                            ) {
                                                state.sessions.broadcast(message);
                                            }
                                        }
                                    }
                                }
                            }
                            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                                tracing::warn!(skipped, "HTTP notification fan-out lagged");
                                if let Some(message) = notification_to_sse_message(
                                    ProtocolNotification::LoggingMessage {
                                        params: ProtocolNotification::control_lagged_logging_params(
                                            skipped,
                                            "http",
                                        ),
                                    },
                                ) {
                                    state.sessions.broadcast(message);
                                }
                            }
                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                        }
                    }
                }
            }
        });

        // Separate logging fan-out task (isolated from control notifications)
        let log_state = Arc::clone(self);
        let mut log_rx = log_state.router.subscribe_logging();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    biased;
                    _ = log_state.cancel.cancelled() => break,
                    recv = log_rx.recv() => {
                        match recv {
                            Ok(notif @ ProtocolNotification::LoggingMessage { .. }) => {
                                if let Some(message) = notification_to_sse_message(notif) {
                                    log_state.sessions.broadcast(message);
                                }
                            }
                            Ok(_) => {} // non-logging notifications on wrong channel
                            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                                tracing::warn!(skipped, "HTTP logging fan-out lagged");
                                // Emit synthetic warning to all connected clients
                                let synthetic = ProtocolNotification::LoggingMessage {
                                    params: rmcp::model::LoggingMessageNotificationParam {
                                        level: rmcp::model::LoggingLevel::Warning,
                                        logger: Some("plug".to_string()),
                                        data: serde_json::json!(format!(
                                            "skipped {skipped} log messages"
                                        )),
                                    },
                                };
                                if let Some(message) = notification_to_sse_message(synthetic) {
                                    log_state.sessions.broadcast(message);
                                }
                            }
                            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                        }
                    }
                }
            }
        });
    }
}

fn notification_to_sse_message(notification: ProtocolNotification) -> Option<SseMessage> {
    SseMessage::from_json_value(notification.to_json_value())
        .map_err(|error| {
            tracing::error!(%error, "failed to serialize SSE notification payload");
        })
        .ok()
}

/// Build the axum Router with all middleware and handlers.
pub fn build_router(state: Arc<HttpState>) -> Router {
    state.spawn_notification_fanout();

    // Discovery endpoint — exempt from origin validation
    let discovery = Router::new()
        .route("/.well-known/mcp.json", get(get_server_card))
        .route("/.well-known/mcp-server-card", get(get_server_card))
        .route(
            "/.well-known/oauth-authorization-server",
            get(get_oauth_authorization_server_metadata),
        )
        .route(
            "/.well-known/oauth-protected-resource",
            get(get_oauth_protected_resource_metadata),
        )
        .route(
            "/.well-known/oauth-protected-resource/mcp",
            get(get_oauth_protected_resource_metadata),
        )
        .route("/oauth/authorize", get(oauth_authorize_not_implemented))
        .route("/oauth/token", post(oauth_token_not_implemented))
        .with_state(state.clone());

    // MCP protocol routes — protected by auth + origin validation middleware
    // Layer order (innermost first): origin validation → bearer auth → body limit
    // Bearer auth runs first; if authenticated, origin validation is skipped.
    let mcp = Router::new()
        .route("/mcp", post(post_mcp).get(get_mcp).delete(delete_mcp))
        .layer(middleware::from_fn_with_state(
            state.clone(),
            validate_origin,
        ))
        .layer(middleware::from_fn_with_state(
            state.clone(),
            validate_bearer_auth,
        ))
        .layer(RequestBodyLimitLayer::new(4 * 1024 * 1024)) // 4MB DoS prevention
        .with_state(state);

    discovery.merge(mcp)
}

// ---------------------------------------------------------------------------
// Middleware
// ---------------------------------------------------------------------------

/// Whether the request has been authenticated via bearer token.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AuthStatus {
    /// Request authenticated with valid bearer token.
    Authenticated,
    /// No auth required (loopback-only server).
    NoAuthRequired,
}

/// Check if a request's bearer token is valid against the expected token.
fn check_bearer_token(headers: &HeaderMap, expected: &str) -> bool {
    headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .is_some_and(|token| crate::auth::verify_auth_token(token, expected))
}

#[derive(Debug, serde::Deserialize)]
struct OAuthAuthorizeParams {
    response_type: String,
    client_id: String,
    redirect_uri: String,
    state: String,
    code_challenge: String,
    code_challenge_method: String,
    scope: Option<String>,
}

/// Validate bearer token for non-loopback HTTP servers.
///
/// When `HttpState.auth_token` is `Some`, requests must include a valid
/// `Authorization: Bearer <token>` header. When `None`, all requests pass through.
async fn validate_bearer_auth(
    State(state): State<Arc<HttpState>>,
    mut req: Request,
    next: Next,
) -> Result<Response, HttpError> {
    if state.auth_mode == crate::config::DownstreamAuthMode::Oauth {
        let manager = state
            .downstream_oauth
            .as_ref()
            .ok_or_else(|| HttpError::Internal("missing downstream OAuth manager".to_string()))?;
        let metadata_url = protected_resource_metadata_url(manager.base_url());
        let scopes = resource_scopes(&manager.config.oauth_scopes);
        let scope = (!scopes.is_empty()).then(|| scopes.join(" "));

        let auth_status = if let Some(auth_header) = req.headers().get(header::AUTHORIZATION) {
            let auth = auth_header
                .to_str()
                .map_err(|_| HttpError::UnauthorizedWithMetadata {
                    metadata_url: metadata_url.clone(),
                    scope: scope.clone(),
                })?
                .strip_prefix("Bearer ")
                .ok_or(HttpError::UnauthorizedWithMetadata {
                    metadata_url: metadata_url.clone(),
                    scope: scope.clone(),
                })?;
            if manager.validate_access_token(auth).await {
                AuthStatus::Authenticated
            } else {
                return Err(HttpError::UnauthorizedWithMetadata {
                    metadata_url,
                    scope,
                });
            }
        } else {
            return Err(HttpError::UnauthorizedWithMetadata {
                metadata_url,
                scope,
            });
        };

        req.extensions_mut().insert(auth_status);
        return Ok(next.run(req).await);
    }

    let auth_status = match &state.auth_token {
        None => AuthStatus::NoAuthRequired,
        Some(expected) => {
            if check_bearer_token(req.headers(), expected.as_ref()) {
                AuthStatus::Authenticated
            } else {
                tracing::warn!("bearer auth failed from downstream client");
                return Err(HttpError::Unauthorized);
            }
        }
    };

    req.extensions_mut().insert(auth_status);
    Ok(next.run(req).await)
}

/// Validate Origin header for DNS rebinding prevention.
///
/// - Missing Origin: allowed (non-browser MCP clients don't send it)
/// - localhost/127.0.0.1/[::1]: allowed
/// - "null" literal: rejected (DNS rebinding vector)
/// - Anything else: rejected
/// - Authenticated requests (via bearer token): origin check skipped
async fn validate_origin(
    State(state): State<Arc<HttpState>>,
    req: Request,
    next: Next,
) -> Result<Response, HttpError> {
    // Skip origin check for authenticated remote clients
    if req.extensions().get::<AuthStatus>() == Some(&AuthStatus::Authenticated) {
        return Ok(next.run(req).await);
    }

    if let Some(origin) = req.headers().get(header::ORIGIN) {
        let origin = origin.to_str().map_err(|_| HttpError::InvalidOrigin)?;

        if state
            .allowed_origins
            .iter()
            .any(|allowed| allowed.as_ref() == origin)
        {
            return Ok(next.run(req).await);
        }

        if origin == "null" {
            return Err(HttpError::InvalidOrigin);
        }

        // Parse origin to extract host — prevents bypass via localhost.evil.com
        let is_local = if let Some(host) = extract_origin_host(origin) {
            host == "localhost" || host == "127.0.0.1" || host == "[::1]" || host == "::1"
        } else {
            false
        };

        if !is_local {
            return Err(HttpError::InvalidOrigin);
        }
    }

    Ok(next.run(req).await)
}

// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------

/// POST /mcp — handle JSON-RPC requests, notifications, and client responses.
async fn post_mcp(
    State(state): State<Arc<HttpState>>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, HttpError> {
    // 1. Validate Content-Type
    let content_type = headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    if !content_type.starts_with("application/json") {
        return Err(HttpError::InvalidContentType);
    }

    // 2. Parse JSON-RPC message
    let message: ClientJsonRpcMessage = serde_json::from_slice(&body).map_err(|e| {
        tracing::debug!(error = %e, "invalid JSON-RPC message from client");
        HttpError::BadRequest("invalid JSON-RPC message".into())
    })?;

    validate_protocol_version_for_post(&headers, &message)?;
    if let Err(err) = validate_mirrored_headers(&headers, &message) {
        return Ok(header_mismatch_response(&message, err));
    }
    let trace_id = Arc::<str>::from(extract_trace_id(&headers));

    // 3. Route based on message type
    match message {
        JsonRpcMessage::Request(req) => handle_request(req, &headers, &state, trace_id).await,
        JsonRpcMessage::Response(response) => {
            let session_id = extract_session_id(&headers)?;
            validate_session_header(&headers, state.sessions.as_ref())?;
            handle_client_response(response, &session_id, &state).await?;
            Ok(StatusCode::ACCEPTED.into_response())
        }
        JsonRpcMessage::Notification(notification) => {
            let session_id = extract_session_id(&headers)?;
            validate_session_header(&headers, state.sessions.as_ref())?;
            match notification.notification {
                ClientNotification::CancelledNotification(cancelled) => {
                    state.router.forward_cancel_from_downstream(
                        &DownstreamCallContext::http_for_client_with_trace(
                            Arc::<str>::from(session_id.as_str()),
                            cancelled.params.request_id.clone(),
                            crate::types::ClientType::Unknown,
                            Arc::clone(&trace_id),
                        ),
                        cancelled.params.reason,
                    );
                }
                ClientNotification::InitializedNotification(_) => {
                    maybe_request_http_roots(Arc::clone(&state), session_id.clone());
                    let caps = state
                        .client_capabilities
                        .get(&session_id)
                        .map(|c| c.clone())
                        .unwrap_or_default();
                    let session_arc = Arc::<str>::from(session_id.as_str());
                    let bridge = Arc::new(HttpBridge {
                        state: Arc::clone(&state),
                        session_id: Arc::clone(&session_arc),
                        capabilities: caps,
                    });
                    state.router.register_downstream_bridge(
                        NotificationTarget::Http {
                            session_id: session_arc,
                        },
                        bridge,
                    );
                }
                ClientNotification::RootsListChangedNotification(_) => {
                    maybe_request_http_roots(Arc::clone(&state), session_id.clone());
                }
                _ => {}
            }
            Ok(StatusCode::ACCEPTED.into_response())
        }
        JsonRpcMessage::Error(_) => Err(HttpError::BadRequest(
            "unexpected error message from client".into(),
        )),
    }
}

fn header_mismatch_response(message: &ClientJsonRpcMessage, err: HeaderMismatch) -> Response {
    let id = match message {
        JsonRpcMessage::Request(request) => serde_json::to_value(&request.id).unwrap_or_default(),
        _ => serde_json::Value::Null,
    };
    let body = json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {
            "code": HEADER_MISMATCH_CODE,
            "message": err.message,
        }
    });
    (StatusCode::BAD_REQUEST, Json(body)).into_response()
}

fn validate_protocol_version_for_post(
    headers: &HeaderMap,
    message: &ClientJsonRpcMessage,
) -> Result<(), HttpError> {
    let require_header = !matches!(
        message,
        JsonRpcMessage::Request(req)
            if matches!(req.request, ClientRequest::InitializeRequest(_))
    );

    match headers.get(PROTOCOL_VERSION_HEADER) {
        Some(value) => {
            let version = value
                .to_str()
                .map_err(|_| HttpError::BadRequest("invalid MCP-Protocol-Version header".into()))?;
            if version != PROTOCOL_VERSION {
                return Err(HttpError::UnsupportedProtocolVersion(version.to_string()));
            }
            Ok(())
        }
        None if require_header => Err(HttpError::MissingProtocolVersion),
        None => Ok(()),
    }
}

fn extract_trace_id(headers: &HeaderMap) -> String {
    if let Some(trace_id) = headers
        .get(TRACEPARENT_HEADER)
        .and_then(|value| value.to_str().ok())
        .and_then(trace_id_from_traceparent)
    {
        return trace_id.to_string();
    }

    if let Some(trace_id) = headers
        .get(PLUG_TRACE_ID_HEADER)
        .and_then(|value| value.to_str().ok())
        .filter(|value| is_valid_trace_id(value))
    {
        return trace_id.to_ascii_lowercase();
    }

    crate::proxy::new_trace_id()
}

fn trace_id_from_traceparent(value: &str) -> Option<&str> {
    let mut parts = value.split('-');
    let version = parts.next()?;
    let trace_id = parts.next()?;
    let parent_id = parts.next()?;
    let flags = parts.next()?;
    if parts.next().is_some()
        || version.len() != 2
        || parent_id.len() != 16
        || flags.len() != 2
        || !is_valid_trace_id(trace_id)
    {
        return None;
    }
    Some(trace_id)
}

fn is_valid_trace_id(value: &str) -> bool {
    value.len() == 32
        && value != "00000000000000000000000000000000"
        && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}

/// Send a reverse JSON-RPC request to an HTTP client via its SSE stream and
/// await the response posted back via POST.
async fn send_http_client_request(
    state: &HttpState,
    session_id: &str,
    request: ServerRequest,
    timeout: Option<Duration>,
) -> Result<ClientResult, McpError> {
    let has_live_sse_sender = state
        .sessions
        .has_live_sse_sender(session_id)
        .map_err(|_| {
            McpError::internal_error(
                "HTTP client has no live SSE stream for reverse requests".to_string(),
                None,
            )
        })?;
    if !has_live_sse_sender {
        return Err(McpError::internal_error(
            "HTTP client has no live SSE stream for reverse requests".to_string(),
            None,
        ));
    }

    let id = state.reverse_request_counter.fetch_add(1, Ordering::SeqCst) as i64;
    let request_id = RequestId::from(NumberOrString::Number(id));
    let message = ServerJsonRpcMessage::request(request, request_id);
    let message = serde_json::to_value(message)
        .map_err(|e| McpError::internal_error(e.to_string(), None))
        .and_then(|value| {
            SseMessage::from_json_value_with_replay_key(value, SseReplayKey::ReverseRequest(id))
                .map_err(|e| McpError::internal_error(e.to_string(), None))
        })?;

    let has_live_sse_sender = state
        .sessions
        .has_live_sse_sender(session_id)
        .map_err(|_| {
            McpError::internal_error(
                "HTTP client SSE stream disappeared before reverse request delivery".to_string(),
                None,
            )
        })?;
    if !has_live_sse_sender {
        return Err(McpError::internal_error(
            "HTTP client SSE stream disappeared before reverse request delivery".to_string(),
            None,
        ));
    }

    let (tx, rx) = oneshot::channel();
    state
        .pending_client_requests
        .insert((session_id.to_string(), id), tx);
    match state.sessions.send_to_live_session(session_id, message) {
        SessionSendOutcome::Delivered | SessionSendOutcome::Queued => {}
        SessionSendOutcome::SessionNotFound => {
            state
                .pending_client_requests
                .remove(&(session_id.to_string(), id));
            return Err(McpError::internal_error(
                "HTTP client SSE stream could not accept reverse request delivery".to_string(),
                None,
            ));
        }
    }
    match timeout {
        Some(duration) => match tokio::time::timeout(duration, rx).await {
            Ok(Ok(result)) => {
                state
                    .sessions
                    .remove_replay_events_by_key(session_id, &SseReplayKey::ReverseRequest(id));
                Ok(result)
            }
            Ok(Err(_)) => Err(McpError::internal_error(
                "HTTP client response channel closed".to_string(),
                None,
            )),
            Err(_) => {
                // Clean up the pending request on timeout
                state
                    .pending_client_requests
                    .remove(&(session_id.to_string(), id));
                state
                    .sessions
                    .remove_replay_events_by_key(session_id, &SseReplayKey::ReverseRequest(id));
                Err(McpError::internal_error(
                    "HTTP client request timed out".to_string(),
                    None,
                ))
            }
        },
        None => match rx.await {
            Ok(result) => {
                state
                    .sessions
                    .remove_replay_events_by_key(session_id, &SseReplayKey::ReverseRequest(id));
                Ok(result)
            }
            Err(_) => Err(McpError::internal_error(
                "HTTP client response channel closed".to_string(),
                None,
            )),
        },
    }
}

/// Handle a client response (POST) that is the answer to a reverse request
/// we sent via SSE (e.g. roots/list).
async fn handle_client_response(
    response: JsonRpcResponse<ClientResult>,
    session_id: &str,
    state: &HttpState,
) -> Result<(), HttpError> {
    let request_id = match response.id {
        RequestId::Number(id) => id,
        RequestId::String(_) => return Ok(()),
    };

    if let Some((_, tx)) = state
        .pending_client_requests
        .remove(&(session_id.to_string(), request_id))
    {
        state
            .sessions
            .remove_replay_events_by_key(session_id, &SseReplayKey::ReverseRequest(request_id));
        let _ = tx.send(response.result);
    }

    Ok(())
}

/// If the session supports roots, spawn a task to request roots via SSE
/// and cache the result.
fn maybe_request_http_roots(state: Arc<HttpState>, session_id: String) {
    if !state.roots_capable_sessions.contains_key(&session_id) {
        return;
    }
    tokio::spawn(async move {
        match send_http_client_request(
            &state,
            &session_id,
            ServerRequest::ListRootsRequest(ListRootsRequest {
                method: Default::default(),
                extensions: Default::default(),
            }),
            Some(Duration::from_secs(10)),
        )
        .await
        {
            Ok(ClientResult::ListRootsResult(result)) => {
                let target = NotificationTarget::Http {
                    session_id: Arc::from(session_id.as_str()),
                };
                if state.router.set_roots_for_target(target, result.roots) {
                    state.router.forward_roots_list_changed_to_upstreams().await;
                }
            }
            Ok(_) => {}
            Err(error) => {
                tracing::debug!(
                    error = %error,
                    session_id = %session_id,
                    "failed to refresh HTTP roots"
                );
            }
        }
    });
}

/// GET /mcp — open SSE stream for server-initiated notifications.
async fn get_mcp(
    State(state): State<Arc<HttpState>>,
    headers: HeaderMap,
) -> Result<Response, HttpError> {
    // 1. Validate session
    let session_id = extract_session_id(&headers)?;
    state.sessions.validate(&session_id)?;

    // 2. Validate Accept header
    let accept = headers
        .get(header::ACCEPT)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    if !accept.contains("text/event-stream") {
        return Err(HttpError::InvalidAcceptHeader);
    }

    // 3. Create channel and register SSE sender
    let (tx, rx) = mpsc::channel(state.sse_channel_capacity);
    let last_event_id = headers
        .get("last-event-id")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.trim().parse::<u64>().ok());
    state
        .sessions
        .set_sse_sender(&session_id, tx, last_event_id)?;

    // 4. Build SSE response with appropriate headers
    let session_store = Arc::clone(&state.sessions);
    let keepalive_session_id = session_id.clone();
    let sse = sse_stream_with_heartbeat(rx, state.cancel.clone(), move || {
        let _ = session_store.touch(&keepalive_session_id);
    });
    let mut response = sse.into_response();
    response
        .headers_mut()
        .insert("X-Accel-Buffering", HeaderValue::from_static("no"));
    response
        .headers_mut()
        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
    response.headers_mut().insert(
        "X-Content-Type-Options",
        HeaderValue::from_static("nosniff"),
    );

    Ok(response)
}

/// DELETE /mcp — terminate a session.
async fn delete_mcp(
    State(state): State<Arc<HttpState>>,
    headers: HeaderMap,
) -> Result<Response, HttpError> {
    let session_id = extract_session_id(&headers)?;

    if state.sessions.remove(&session_id) {
        // Clean up resource subscriptions for this departing session
        let target = NotificationTarget::Http {
            session_id: Arc::from(session_id.as_str()),
        };
        state.router.cleanup_subscriptions_for_target(&target).await;
        state.roots_capable_sessions.remove(&session_id);
        state.client_capabilities.remove(&session_id);
        // Drop pending reverse-request senders so receivers get RecvError
        state
            .pending_client_requests
            .retain(|(pending_session_id, _), _| pending_session_id != &session_id);
        state.router.unregister_downstream_bridge(&target);
        if state.router.clear_roots_for_target(&target) {
            state.router.forward_roots_list_changed_to_upstreams().await;
        }
        // Clean up per-client log level to prevent stale entries from
        // keeping the effective level permanently at a permissive value.
        state.router.remove_client_log_level(&session_id);
        let lazy_session_key = crate::proxy::ToolRouter::lazy_session_key(
            crate::proxy::DownstreamTransport::Http,
            &session_id,
        );
        state.router.clear_lazy_session(&lazy_session_key);
        tracing::info!(session_id = %session_id, "session terminated via DELETE");
        Ok(StatusCode::OK.into_response())
    } else {
        Err(HttpError::SessionNotFound)
    }
}

/// GET /.well-known/mcp-server-card and legacy /.well-known/mcp.json.
async fn get_server_card(
    State(state): State<Arc<HttpState>>,
    _headers: HeaderMap,
) -> impl IntoResponse {
    let mut remote = json!({
        "type": "streamable-http",
        "url": "/mcp",
        "supportedProtocolVersions": [PROTOCOL_VERSION],
    });
    if state.auth_mode == crate::config::DownstreamAuthMode::Oauth || state.auth_token.is_some() {
        remote["headers"] = json!([
            {
                "name": "Authorization",
                "description": "Bearer access token",
                "isRequired": true,
                "isSecret": true
            }
        ]);
    };

    let card = json!({
        "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json",
        "name": "io.github.cyberpapiii/plug",
        "version": env!("CARGO_PKG_VERSION"),
        "description": "MCP multiplexer for sharing one server configuration across many AI clients",
        "title": "Plug",
        "websiteUrl": "https://github.com/cyberpapiii/plug",
        "repository": {
            "url": "https://github.com/cyberpapiii/plug",
            "source": "github"
        },
        "remotes": [remote],
    });

    let mut response = (StatusCode::OK, Json(card)).into_response();
    response.headers_mut().insert(
        header::CACHE_CONTROL,
        HeaderValue::from_static("public, max-age=3600"),
    );
    response.headers_mut().insert(
        header::ACCESS_CONTROL_ALLOW_ORIGIN,
        HeaderValue::from_static("*"),
    );
    response.headers_mut().insert(
        header::ACCESS_CONTROL_ALLOW_METHODS,
        HeaderValue::from_static("GET"),
    );
    response.headers_mut().insert(
        header::ACCESS_CONTROL_ALLOW_HEADERS,
        HeaderValue::from_static("Content-Type"),
    );
    response.headers_mut().insert(
        "X-Content-Type-Options",
        HeaderValue::from_static("nosniff"),
    );
    response
}

async fn get_oauth_authorization_server_metadata(
    State(state): State<Arc<HttpState>>,
) -> impl IntoResponse {
    let Some(manager) = &state.downstream_oauth else {
        return StatusCode::NOT_FOUND.into_response();
    };

    let base = manager.config.public_base_url.trim_end_matches('/');
    let token_endpoint_auth_methods_supported = match manager.config.oauth_client_secret {
        Some(_) => vec!["client_secret_basic", "client_secret_post"],
        None => vec!["none"],
    };
    let mut grant_types_supported = vec!["authorization_code", "refresh_token"];
    if manager.config.oauth_client_secret.is_some() {
        grant_types_supported.push("client_credentials");
    }
    let response = Json(json!({
        "issuer": base,
        "authorization_endpoint": format!("{base}/oauth/authorize"),
        "token_endpoint": format!("{base}/oauth/token"),
        "response_types_supported": ["code"],
        "grant_types_supported": grant_types_supported,
        "code_challenge_methods_supported": ["S256"],
        "token_endpoint_auth_methods_supported": token_endpoint_auth_methods_supported,
        "scopes_supported": manager.config.oauth_scopes,
    }));
    (StatusCode::OK, response).into_response()
}

async fn get_oauth_protected_resource_metadata(
    State(state): State<Arc<HttpState>>,
) -> impl IntoResponse {
    let Some(manager) = &state.downstream_oauth else {
        return StatusCode::NOT_FOUND.into_response();
    };

    let base = manager.base_url();
    let response = Json(json!({
        "resource": format!("{base}/mcp"),
        "authorization_servers": [base],
        "scopes_supported": resource_scopes(&manager.config.oauth_scopes),
        "bearer_methods_supported": ["header"],
    }));
    (StatusCode::OK, response).into_response()
}

async fn oauth_authorize_not_implemented(
    State(state): State<Arc<HttpState>>,
    Query(params): Query<OAuthAuthorizeParams>,
) -> impl IntoResponse {
    let Some(manager) = &state.downstream_oauth else {
        return StatusCode::NOT_FOUND.into_response();
    };

    if params.response_type != "code" {
        return (StatusCode::BAD_REQUEST, "unsupported response_type").into_response();
    }

    match manager
        .build_authorize_redirect(
            &params.client_id,
            &params.redirect_uri,
            &params.state,
            &params.code_challenge,
            &params.code_challenge_method,
            params.scope.as_deref(),
        )
        .await
    {
        Ok(location) => {
            let mut response = StatusCode::FOUND.into_response();
            if let Ok(value) = HeaderValue::from_str(&location) {
                response.headers_mut().insert(header::LOCATION, value);
                response
            } else {
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "invalid redirect location generated",
                )
                    .into_response()
            }
        }
        Err(error) => (
            StatusCode::BAD_REQUEST,
            format!("authorization request rejected: {error}"),
        )
            .into_response(),
    }
}

async fn oauth_token_not_implemented(
    State(state): State<Arc<HttpState>>,
    headers: HeaderMap,
    Form(params): Form<HashMap<String, String>>,
) -> impl IntoResponse {
    let Some(manager) = &state.downstream_oauth else {
        return StatusCode::NOT_FOUND.into_response();
    };

    let grant_type = match params.get("grant_type").map(String::as_str) {
        Some(value) => value,
        None => return (StatusCode::BAD_REQUEST, "missing grant_type").into_response(),
    };

    let (client_id, client_secret) = match manager.client_credentials_from_headers_or_form(
        &headers,
        params.get("client_id").map(String::as_str),
        params.get("client_secret").map(String::as_str),
    ) {
        Ok(credentials) => credentials,
        Err(error) => {
            return (
                StatusCode::UNAUTHORIZED,
                format!("client authentication failed: {error}"),
            )
                .into_response();
        }
    };

    let result = match grant_type {
        "authorization_code" => {
            let code = match params.get("code") {
                Some(value) => value,
                None => return (StatusCode::BAD_REQUEST, "missing code").into_response(),
            };
            let redirect_uri = match params.get("redirect_uri") {
                Some(value) => value,
                None => return (StatusCode::BAD_REQUEST, "missing redirect_uri").into_response(),
            };
            let code_verifier = match params.get("code_verifier") {
                Some(value) => value,
                None => return (StatusCode::BAD_REQUEST, "missing code_verifier").into_response(),
            };
            manager
                .exchange_authorization_code(
                    &client_id,
                    client_secret.as_deref(),
                    code,
                    redirect_uri,
                    code_verifier,
                )
                .await
        }
        "refresh_token" => {
            let refresh_token = match params.get("refresh_token") {
                Some(value) => value,
                None => return (StatusCode::BAD_REQUEST, "missing refresh_token").into_response(),
            };
            manager
                .exchange_refresh_token(&client_id, client_secret.as_deref(), refresh_token)
                .await
        }
        "client_credentials" => {
            manager
                .exchange_client_credentials(
                    &client_id,
                    client_secret.as_deref(),
                    params.get("scope").map(String::as_str),
                )
                .await
        }
        _ => return (StatusCode::BAD_REQUEST, "unsupported grant_type").into_response(),
    };

    match result {
        Ok(token) => {
            let mut body = json!({
                "access_token": token.access_token,
                "token_type": "Bearer",
                "expires_in": token.expires_in,
                "scope": token.scope,
            });
            if let Some(refresh_token) = token.refresh_token {
                body["refresh_token"] = json!(refresh_token);
            }
            (StatusCode::OK, Json(body)).into_response()
        }
        Err(error) => (
            StatusCode::BAD_REQUEST,
            format!("token exchange failed: {error}"),
        )
            .into_response(),
    }
}

fn protected_resource_metadata_url(base_url: &str) -> String {
    format!(
        "{}/.well-known/oauth-protected-resource",
        base_url.trim_end_matches('/')
    )
}

// ---------------------------------------------------------------------------
// Request routing
// ---------------------------------------------------------------------------

/// Route a typed JSON-RPC request to the appropriate handler.
async fn handle_request(
    req: JsonRpcRequest<ClientRequest>,
    headers: &HeaderMap,
    state: &Arc<HttpState>,
    trace_id: Arc<str>,
) -> Result<Response, HttpError> {
    let request_id = req.id.clone();

    match req.request {
        ClientRequest::InitializeRequest(init_req) => {
            // Initialize: create session, return server info
            let session_id = state.sessions.create_session()?;

            let client_name = &init_req.params.client_info.name;
            let client_type = crate::client_detect::detect_client(client_name);
            tracing::info!(
                client = %client_name,
                detected = %client_type,
                session = %session_id,
                trace_id = %trace_id,
                "HTTP client connected"
            );
            // Store client type in session
            let _ = state.sessions.set_client_type(&session_id, client_type);

            // Track roots capability for reverse-request roots fetching
            if init_req.params.capabilities.roots.is_some() {
                state.roots_capable_sessions.insert(session_id.clone(), ());
            }

            // Store client capabilities for bridge capability gating
            state
                .client_capabilities
                .insert(session_id.clone(), init_req.params.capabilities.clone());

            let result = build_initialize_result(state.router.as_ref(), client_type);

            let response_msg =
                ServerJsonRpcMessage::response(ServerResult::InitializeResult(result), request_id);

            json_response_with_session(&session_id, &response_msg)
        }

        ClientRequest::PingRequest(_) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            let response_msg = ServerJsonRpcMessage::response(
                ServerResult::EmptyResult(EmptyResult {}),
                request_id,
            );
            json_response(&response_msg)
        }

        ClientRequest::ListToolsRequest(list_req) => {
            let session_id_str = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let client_type = state
                .sessions
                .get_client_type(&session_id_str)
                .unwrap_or(crate::types::ClientType::Unknown);
            let lazy_session_key = crate::proxy::ToolRouter::lazy_session_key(
                crate::proxy::DownstreamTransport::Http,
                &session_id_str,
            );
            let result = state.router.list_tools_page_for_client_session(
                client_type,
                Some(&lazy_session_key),
                list_req.params,
            );
            let response_msg =
                ServerJsonRpcMessage::response(ServerResult::ListToolsResult(result), request_id);
            json_response(&response_msg)
        }

        ClientRequest::CallToolRequest(call_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let client_type = state
                .sessions
                .get_client_type(&session_id)
                .unwrap_or(crate::types::ClientType::Unknown);
            let progress_token = call_req.params.progress_token();
            let owner =
                crate::tasks::TaskOwner::new(Arc::<str>::from(format!("http:{session_id}")));
            let downstream = DownstreamCallContext::http_for_client_with_trace(
                Arc::<str>::from(session_id.as_str()),
                request_id.clone(),
                client_type,
                Arc::clone(&trace_id),
            );
            if call_req.params.task.is_some() {
                match state
                    .router
                    .clone()
                    .enqueue_tool_task(
                        call_req.params.name.as_ref(),
                        call_req.params.arguments,
                        progress_token,
                        owner,
                        Some(downstream),
                    )
                    .await
                {
                    Ok(result) => {
                        let response_msg = ServerJsonRpcMessage::response(
                            ServerResult::CreateTaskResult(result),
                            request_id,
                        );
                        json_response(&response_msg)
                    }
                    Err(mcp_err) => {
                        let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                        json_response(&response_msg)
                    }
                }
            } else {
                match state
                    .router
                    .call_tool_with_context(
                        call_req.params.name.as_ref(),
                        call_req.params.arguments,
                        progress_token,
                        Some(downstream),
                    )
                    .await
                {
                    Ok(result) => {
                        let response_msg = ServerJsonRpcMessage::response(
                            ServerResult::CallToolResult(result),
                            request_id,
                        );
                        json_response(&response_msg)
                    }
                    Err(mcp_err) => {
                        let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                        json_response(&response_msg)
                    }
                }
            }
        }

        ClientRequest::ListTasksRequest(list_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let owner =
                crate::tasks::TaskOwner::new(Arc::<str>::from(format!("http:{session_id}")));
            match state
                .router
                .list_tasks_for_owner(&owner, list_req.params)
                .await
            {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::ListTasksResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::GetTaskInfoRequest(task_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let owner =
                crate::tasks::TaskOwner::new(Arc::<str>::from(format!("http:{session_id}")));
            match state
                .router
                .get_task_info_for_owner(&owner, &task_req.params.task_id)
                .await
            {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::GetTaskResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::GetTaskResultRequest(task_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let owner =
                crate::tasks::TaskOwner::new(Arc::<str>::from(format!("http:{session_id}")));
            match state
                .router
                .get_task_result_for_owner(&owner, &task_req.params.task_id)
                .await
            {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::GetTaskPayloadResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::CancelTaskRequest(task_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let owner =
                crate::tasks::TaskOwner::new(Arc::<str>::from(format!("http:{session_id}")));
            match state
                .router
                .cancel_task_for_owner(&owner, &task_req.params.task_id)
                .await
            {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::CancelTaskResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::ListResourcesRequest(list_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            let result = state.router.list_resources_page(list_req.params);
            let response_msg = ServerJsonRpcMessage::response(
                ServerResult::ListResourcesResult(result),
                request_id,
            );
            json_response(&response_msg)
        }

        ClientRequest::ListResourceTemplatesRequest(list_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            let result = state.router.list_resource_templates_page(list_req.params);
            let response_msg = ServerJsonRpcMessage::response(
                ServerResult::ListResourceTemplatesResult(result),
                request_id,
            );
            json_response(&response_msg)
        }

        ClientRequest::ReadResourceRequest(read_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            match state.router.read_resource(&read_req.params.uri).await {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::ReadResourceResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::ListPromptsRequest(list_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            let result = state.router.list_prompts_page(list_req.params);
            let response_msg =
                ServerJsonRpcMessage::response(ServerResult::ListPromptsResult(result), request_id);
            json_response(&response_msg)
        }

        ClientRequest::GetPromptRequest(prompt_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            match state
                .router
                .get_prompt(&prompt_req.params.name, prompt_req.params.arguments)
                .await
            {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::GetPromptResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::SubscribeRequest(sub_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let target = NotificationTarget::Http {
                session_id: Arc::from(session_id.as_str()),
            };
            match state
                .router
                .subscribe_resource(&sub_req.params.uri, target)
                .await
            {
                Ok(()) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::EmptyResult(().into()),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::UnsubscribeRequest(unsub_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            let target = NotificationTarget::Http {
                session_id: Arc::from(session_id.as_str()),
            };
            match state
                .router
                .unsubscribe_resource(&unsub_req.params.uri, &target)
                .await
            {
                Ok(()) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::EmptyResult(().into()),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::CompleteRequest(complete_req) => {
            validate_session_header(headers, state.sessions.as_ref())?;
            match state.router.complete_request(complete_req.params).await {
                Ok(result) => {
                    let response_msg = ServerJsonRpcMessage::response(
                        ServerResult::CompleteResult(result),
                        request_id,
                    );
                    json_response(&response_msg)
                }
                Err(mcp_err) => {
                    let response_msg = ServerJsonRpcMessage::error(mcp_err, Some(request_id));
                    json_response(&response_msg)
                }
            }
        }

        ClientRequest::SetLevelRequest(set_level_req) => {
            let session_id = extract_session_id(headers)?;
            validate_session_header(headers, state.sessions.as_ref())?;
            tracing::info!(
                session = %session_id,
                level = ?set_level_req.params.level,
                "HTTP client set log level"
            );
            state
                .router
                .set_client_log_level(&session_id, set_level_req.params.level);
            state.router.forward_set_level_to_upstreams().await;
            let response_msg = ServerJsonRpcMessage::response(
                ServerResult::EmptyResult(EmptyResult {}),
                request_id,
            );
            json_response(&response_msg)
        }

        _ => {
            // Unsupported method — return JSON-RPC method not found error
            validate_session_header(headers, state.sessions.as_ref())?;
            let error = ErrorData::new(ErrorCode::METHOD_NOT_FOUND, "method not supported", None);
            let response_msg = ServerJsonRpcMessage::error(error, Some(request_id));
            json_response(&response_msg)
        }
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Build the InitializeResult (same as ProxyHandler::get_info).
fn build_initialize_result(
    router: &ToolRouter,
    client_type: crate::types::ClientType,
) -> InitializeResult {
    InitializeResult::new(router.synthesized_capabilities_for_client(client_type))
        .with_server_info(
            Implementation::new("plug", env!("CARGO_PKG_VERSION"))
                .with_title("Plug")
                .with_description("MCP multiplexer")
                .with_website_url("https://github.com/cyberpapiii/plug")
                .with_icons(vec![Icon::new(
                    "https://raw.githubusercontent.com/cyberpapiii/plug/main/docs/assets/plug-icon.svg",
                )
                .with_mime_type("image/svg+xml")
                .with_sizes(vec!["any".to_string()])]),
        )
        .with_protocol_version(ProtocolVersion::V_2025_11_25)
}

/// Extract the host from an Origin header value.
///
/// Origin format is `scheme://host[:port]`. We parse manually to avoid
/// pulling in the `url` crate dependency for this single use case.
fn extract_origin_host(origin: &str) -> Option<&str> {
    let after_scheme = origin.split("://").nth(1)?;
    // Handle IPv6 bracket notation: [::1]:port
    if after_scheme.starts_with('[') {
        let end = after_scheme.find(']')?;
        let host = &after_scheme[..=end]; // includes brackets
        return if host.len() > 2 { Some(host) } else { None };
    }
    // Strip port and path: take everything before ':' or '/'
    let host = after_scheme
        .split(':')
        .next()
        .unwrap_or(after_scheme)
        .split('/')
        .next()
        .unwrap_or(after_scheme);
    if host.is_empty() { None } else { Some(host) }
}

/// Extract session ID from request headers.
fn extract_session_id(headers: &HeaderMap) -> Result<String, HttpError> {
    headers
        .get(SESSION_ID_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string())
        .ok_or(HttpError::SessionRequired)
}

/// Validate that the session exists and is not expired.
fn validate_session_header(
    headers: &HeaderMap,
    sessions: &dyn SessionStore,
) -> Result<(), HttpError> {
    let session_id = extract_session_id(headers)?;
    sessions.validate(&session_id)
}

/// Build a JSON response from a ServerJsonRpcMessage.
fn json_response(msg: &ServerJsonRpcMessage) -> Result<Response, HttpError> {
    let body = serde_json::to_vec(msg)
        .map_err(|e| HttpError::Internal(format!("failed to serialize response: {e}")))?;

    let mut response = (StatusCode::OK, body).into_response();
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        HeaderValue::from_static("application/json"),
    );
    response.headers_mut().insert(
        "X-Content-Type-Options",
        HeaderValue::from_static("nosniff"),
    );
    response
        .headers_mut()
        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));

    Ok(response)
}

fn json_response_with_session(
    session_id: &str,
    msg: &ServerJsonRpcMessage,
) -> Result<Response, HttpError> {
    let mut response = json_response(msg)?;
    response.headers_mut().insert(
        SESSION_ID_HEADER,
        HeaderValue::from_str(session_id)
            .map_err(|_| HttpError::Internal("invalid session ID".into()))?,
    );
    response.headers_mut().insert(
        PROTOCOL_VERSION_HEADER,
        HeaderValue::from_static(PROTOCOL_VERSION),
    );

    Ok(response)
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use http::Request as HttpRequest;
    use std::time::Duration;
    use tower::ServiceExt;

    #[test]
    fn trace_id_prefers_valid_w3c_traceparent() {
        let mut headers = HeaderMap::new();
        headers.insert(
            TRACEPARENT_HEADER,
            HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
        );

        assert_eq!(
            extract_trace_id(&headers),
            "4bf92f3577b34da6a3ce929d0e0e4736"
        );
    }

    #[test]
    fn trace_id_falls_back_when_header_is_invalid() {
        let mut headers = HeaderMap::new();
        headers.insert(TRACEPARENT_HEADER, HeaderValue::from_static("invalid"));

        let trace_id = extract_trace_id(&headers);
        assert_eq!(trace_id.len(), 32);
        assert_ne!(trace_id, "00000000000000000000000000000000");
    }

    async fn collect_sse_events(body: Body, max_events: usize) -> Vec<String> {
        let mut events = Vec::new();
        let mut stream = body.into_data_stream();
        use futures::StreamExt;

        let timeout = tokio::time::timeout(Duration::from_secs(2), async {
            while let Some(Ok(chunk)) = stream.next().await {
                let text = String::from_utf8_lossy(&chunk).to_string();
                for part in text.split("\n\n") {
                    let trimmed = part.trim();
                    if !trimmed.is_empty() {
                        events.push(trimmed.to_string());
                    }
                }
                if events.len() >= max_events {
                    break;
                }
            }
        });

        let _ = timeout.await;
        events
    }

    fn sse_event_id(event: &str) -> Option<u64> {
        event.lines().find_map(|line| {
            line.strip_prefix("id: ")
                .and_then(|value| value.trim().parse::<u64>().ok())
        })
    }

    fn test_state_with_router_config(router_config: crate::proxy::RouterConfig) -> Arc<HttpState> {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(sm, router_config));
        Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Auto,
            downstream_oauth: None,
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        })
    }

    fn test_state() -> Arc<HttpState> {
        test_state_with_router_config(crate::proxy::RouterConfig {
            prefix_delimiter: "__".to_string(),
            priority_tools: Vec::new(),
            disabled_tools: Vec::new(),
            tool_description_max_chars: None,
            tool_search_threshold: 50,
            meta_tool_mode: false,
            lazy_tools: crate::config::LazyToolsConfig::default(),
            tool_filter_enabled: true,
            enrichment_servers: std::collections::HashSet::new(),
        })
    }

    #[tokio::test]
    async fn post_without_content_type_returns_415() {
        let app = build_router(test_state());
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
    }

    #[tokio::test]
    async fn post_initialize_returns_session_id() {
        let state = test_state();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-05",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        assert!(resp.headers().get(SESSION_ID_HEADER).is_some());
        assert!(resp.headers().get(PROTOCOL_VERSION_HEADER).is_some());
        assert_eq!(
            resp.headers()
                .get(PROTOCOL_VERSION_HEADER)
                .and_then(|value| value.to_str().ok()),
            Some(PROTOCOL_VERSION)
        );
    }

    #[tokio::test]
    async fn post_rejects_mismatched_mcp_method_header() {
        let app = build_router(test_state());

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 9,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header("Mcp-Method", "tools/list")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let value: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(value["error"]["code"], HEADER_MISMATCH_CODE);
        assert_eq!(value["id"], 9);
        assert!(
            value["error"]["message"]
                .as_str()
                .unwrap()
                .contains("Mcp-Method")
        );
    }

    #[tokio::test]
    async fn post_rejects_mismatched_mcp_name_header() {
        let state = test_state();
        let app = build_router(state.clone());

        let init_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": { "name": "header-test", "version": "1.0" }
            }
        });
        let init_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&init_body).unwrap()))
            .unwrap();
        let init_resp = app.oneshot(init_req).await.unwrap();
        let session_id = init_resp
            .headers()
            .get(SESSION_ID_HEADER)
            .unwrap()
            .to_str()
            .unwrap()
            .to_string();

        let app = build_router(state);
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "actual_tool",
                "arguments": {}
            }
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .header("Mcp-Method", "tools/call")
            .header("Mcp-Name", "spoofed_tool")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let value: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(value["error"]["code"], HEADER_MISMATCH_CODE);
        assert_eq!(value["id"], 2);
        assert!(
            value["error"]["message"]
                .as_str()
                .unwrap()
                .contains("Mcp-Name")
        );
    }

    #[tokio::test]
    async fn post_tools_list_without_session_returns_400() {
        let app = build_router(test_state());

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list"
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn delete_without_session_returns_400() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("DELETE")
            .uri("/mcp")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn delete_with_valid_session_returns_200() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("DELETE")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn get_without_accept_header_returns_406() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_ACCEPTABLE);
    }

    #[tokio::test]
    async fn origin_null_rejected() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "null")
            .header("content-type", "application/json")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn origin_external_rejected() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "https://evil.example.com")
            .header("content-type", "application/json")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn origin_allowlisted_external_origin_accepted() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Auto,
            downstream_oauth: None,
            sse_channel_capacity: 32,
            allowed_origins: vec![Arc::from("https://claude.ai")],
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "https://claude.ai")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    /// Full session lifecycle: initialize → tools/list → ping → delete
    #[tokio::test]
    async fn full_session_lifecycle() {
        let state = test_state();

        // 1. Initialize — get session ID
        let app = build_router(state.clone());
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-05",
                "capabilities": {},
                "clientInfo": { "name": "lifecycle-test", "version": "1.0" }
            }
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let session_id = resp
            .headers()
            .get(SESSION_ID_HEADER)
            .unwrap()
            .to_str()
            .unwrap()
            .to_string();

        // 2. tools/list with session
        let app = build_router(state.clone());
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        // 3. ping with session
        let app = build_router(state.clone());
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "ping"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        // 4. DELETE session
        let app = build_router(state.clone());
        let req = HttpRequest::builder()
            .method("DELETE")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        // 5. Verify session is gone — tools/list should fail
        let app = build_router(state);
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 4,
            "method": "tools/list"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn routed_http_error_preserves_request_id() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": "request-123",
            "method": "resources/read",
            "params": {
                "uri": "memory://missing"
            }
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(json["id"], "request-123");
        assert!(json["error"]["code"].is_number());
    }

    #[tokio::test]
    async fn http_tools_list_uses_meta_tool_mode_surface() {
        let state = test_state_with_router_config(crate::proxy::RouterConfig {
            prefix_delimiter: "__".to_string(),
            priority_tools: Vec::new(),
            disabled_tools: Vec::new(),
            tool_description_max_chars: None,
            tool_search_threshold: 50,
            meta_tool_mode: true,
            lazy_tools: crate::config::LazyToolsConfig::default(),
            tool_filter_enabled: true,
            enrichment_servers: std::collections::HashSet::new(),
        });

        state.router.replace_snapshot(crate::proxy::RouterSnapshot {
            routes: std::collections::HashMap::from([(
                "Git__commit".to_string(),
                ("git".to_string(), "commit".to_string()),
            )]),
            tools_all: Arc::new(vec![Tool::new(
                std::borrow::Cow::Borrowed("Git__commit"),
                std::borrow::Cow::Borrowed("Create a git commit"),
                Arc::new(serde_json::Map::new()),
            )]),
            meta_tools_all: Arc::new(vec![Tool::new(
                std::borrow::Cow::Borrowed("plug__search_tools"),
                std::borrow::Cow::Borrowed("Search tools"),
                Arc::new(serde_json::Map::new()),
            )]),
            tools_windsurf: Arc::new(Vec::new()),
            tools_copilot: Arc::new(Vec::new()),
            resources_all: Arc::new(Vec::new()),
            resource_templates_all: Arc::new(Vec::new()),
            prompts_all: Arc::new(Vec::new()),
            resource_routes: std::collections::HashMap::new(),
            prompt_routes: std::collections::HashMap::new(),
            tool_definition_fingerprints: std::collections::HashMap::new(),
            tool_risk_inventory: std::collections::HashMap::new(),
        });

        let app = build_router(state.clone());
        let session_id = state.sessions.create_session().unwrap();
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
        let names = json["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .map(|tool| tool["name"].as_str().unwrap().to_string())
            .collect::<Vec<_>>();
        assert_eq!(
            names,
            vec![
                "plug__list_servers",
                "plug__list_tools",
                "plug__search_tools",
                "plug__invoke_tool",
            ]
        );
    }

    #[tokio::test]
    async fn notification_without_session_returns_400() {
        let app = build_router(test_state());
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/initialized"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn notification_with_valid_session_returns_202() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/initialized"
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .header(SESSION_ID_HEADER, &session_id)
            .header(PROTOCOL_VERSION_HEADER, PROTOCOL_VERSION)
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
    }

    #[tokio::test]
    async fn initialize_response_contains_server_info() {
        let state = test_state();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-05",
                "capabilities": {},
                "clientInfo": { "name": "info-test", "version": "1.0" }
            }
        });
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let resp_body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&resp_body).unwrap();
        assert_eq!(json["result"]["protocolVersion"], "2025-11-25");
        assert_eq!(json["result"]["serverInfo"]["name"], "plug");
        assert!(json["result"]["capabilities"]["tools"].is_null());
    }

    #[tokio::test]
    async fn origin_localhost_subdomain_rejected() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "http://localhost.evil.com")
            .header("content-type", "application/json")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn server_card_returns_json() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp-server-card")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(
            resp.headers()
                .get(header::CACHE_CONTROL)
                .and_then(|v| v.to_str().ok()),
            Some("public, max-age=3600")
        );
        assert_eq!(
            resp.headers()
                .get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
                .and_then(|v| v.to_str().ok()),
            Some("*")
        );
        assert_eq!(
            resp.headers()
                .get(header::ACCESS_CONTROL_ALLOW_METHODS)
                .and_then(|v| v.to_str().ok()),
            Some("GET")
        );
        assert_eq!(
            resp.headers()
                .get("X-Content-Type-Options")
                .and_then(|v| v.to_str().ok()),
            Some("nosniff")
        );

        let body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(
            json["$schema"],
            "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json"
        );
        assert_eq!(json["name"], "io.github.cyberpapiii/plug");
        assert_eq!(json["title"], "Plug");
        assert!(json["version"].is_string());
        assert_eq!(json["remotes"][0]["type"], "streamable-http");
        assert_eq!(json["remotes"][0]["url"], "/mcp");
        assert_eq!(
            json["remotes"][0]["supportedProtocolVersions"],
            serde_json::json!([PROTOCOL_VERSION])
        );
        assert!(json.get("tools").is_none());
        assert!(json.get("servers").is_none());
    }

    #[tokio::test]
    async fn legacy_server_card_path_returns_same_card() {
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp.json")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), 1024 * 1024)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
        assert_eq!(json["name"], "io.github.cyberpapiii/plug");
        assert_eq!(json["remotes"][0]["url"], "/mcp");
    }

    #[tokio::test]
    async fn server_card_accessible_with_external_origin() {
        // Discovery endpoint must NOT be blocked by origin validation
        let app = build_router(test_state());

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp.json")
            .header("origin", "https://evil.example.com")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn reverse_request_fails_fast_without_live_sse_sender() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();

        let result = send_http_client_request(
            &state,
            &session_id,
            ServerRequest::ListRootsRequest(ListRootsRequest {
                method: Default::default(),
                extensions: Default::default(),
            }),
            Some(Duration::from_millis(50)),
        )
        .await;

        assert!(result.is_err());
        assert!(state.pending_client_requests.is_empty());
    }

    #[tokio::test]
    async fn reverse_request_succeeds_with_live_sse_sender() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let (tx, mut rx) = mpsc::channel(1);
        state
            .sessions
            .set_sse_sender(&session_id, tx, None)
            .unwrap();

        let state_for_response = Arc::clone(&state);
        let session_id_for_response = session_id.clone();
        tokio::spawn(async move {
            let message = rx.recv().await.expect("reverse request message");
            let message: ServerJsonRpcMessage = serde_json::from_str(message.message.as_str())
                .expect("deserialize reverse request");
            let request_id = match message {
                ServerJsonRpcMessage::Request(request) => request.id,
                other => panic!("unexpected reverse request payload: {other:?}"),
            };

            handle_client_response(
                JsonRpcResponse {
                    jsonrpc: Default::default(),
                    id: request_id,
                    result: ClientResult::ListRootsResult(ListRootsResult::default()),
                },
                &session_id_for_response,
                &state_for_response,
            )
            .await
            .expect("post reverse request response");
        });

        let result = send_http_client_request(
            &state,
            &session_id,
            ServerRequest::ListRootsRequest(ListRootsRequest {
                method: Default::default(),
                extensions: Default::default(),
            }),
            Some(Duration::from_secs(1)),
        )
        .await
        .expect("reverse request should succeed");

        match result {
            ClientResult::ListRootsResult(_) => {}
            other => panic!("unexpected reverse request result: {other:?}"),
        }
        assert!(state.pending_client_requests.is_empty());
    }

    #[tokio::test]
    async fn queued_reverse_request_replays_on_reconnect() {
        let state = test_state();
        let session_id = state.sessions.create_session().unwrap();
        let (blocked_tx, _blocked_rx) = mpsc::channel(1);
        blocked_tx
            .try_send(crate::session::SseEvent {
                id: 999,
                message: SseMessage::from_json_value(serde_json::json!({"blocked": true})).unwrap(),
            })
            .unwrap();
        state
            .sessions
            .set_sse_sender(&session_id, blocked_tx, None)
            .unwrap();

        let state_for_request = Arc::clone(&state);
        let session_id_for_request = session_id.clone();
        let pending = tokio::spawn(async move {
            send_http_client_request(
                &state_for_request,
                &session_id_for_request,
                ServerRequest::ListRootsRequest(ListRootsRequest {
                    method: Default::default(),
                    extensions: Default::default(),
                }),
                Some(Duration::from_secs(1)),
            )
            .await
        });

        tokio::time::sleep(Duration::from_millis(20)).await;
        let (reconnect_tx, mut reconnect_rx) = mpsc::channel(4);
        state
            .sessions
            .set_sse_sender(&session_id, reconnect_tx, Some(0))
            .unwrap();

        let event = reconnect_rx.recv().await.expect("replayed reverse request");
        let message: ServerJsonRpcMessage =
            serde_json::from_str(event.message.as_str()).expect("deserialize reverse request");
        let request_id = match message {
            ServerJsonRpcMessage::Request(request) => request.id,
            other => panic!("unexpected reverse request payload: {other:?}"),
        };

        handle_client_response(
            JsonRpcResponse {
                jsonrpc: Default::default(),
                id: request_id,
                result: ClientResult::ListRootsResult(ListRootsResult::default()),
            },
            &session_id,
            &state,
        )
        .await
        .expect("post reverse request response");

        let result = pending
            .await
            .expect("request task")
            .expect("reverse request should complete after reconnect");
        match result {
            ClientResult::ListRootsResult(_) => {}
            other => panic!("unexpected reverse request result: {other:?}"),
        }
        assert!(state.pending_client_requests.is_empty());
    }

    #[tokio::test]
    async fn origin_localhost_accepted() {
        let state = test_state();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-05",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "http://localhost:3282")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn origin_ipv6_localhost_accepted() {
        let state = test_state();
        let app = build_router(state);

        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-05",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("origin", "http://[::1]:3282")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&body).unwrap()))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn tools_list_changed_reaches_http_sse_client() {
        let state = test_state();
        let app = build_router(state.clone());

        let init_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let init_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&init_body).unwrap()))
            .unwrap();

        let init_resp = app.clone().oneshot(init_req).await.unwrap();
        let session_id = init_resp
            .headers()
            .get(SESSION_ID_HEADER)
            .and_then(|value| value.to_str().ok())
            .expect("session id header")
            .to_string();

        let sse_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, session_id)
            .header("accept", "text/event-stream")
            .body(Body::empty())
            .unwrap();

        let sse_resp = app.oneshot(sse_req).await.unwrap();
        assert_eq!(sse_resp.status(), StatusCode::OK);
        let body = sse_resp.into_body();

        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::ToolListChanged,
        );

        let events = collect_sse_events(body, 3).await;
        assert!(
            events
                .iter()
                .any(|event| event.contains("notifications/tools/list_changed")),
            "expected SSE stream to contain tools/list_changed notification, got {events:?}"
        );
    }

    #[tokio::test]
    async fn last_event_id_replays_missed_http_sse_notifications() {
        let state = test_state();
        let app = build_router(state.clone());

        let session_id = state.sessions.create_session().unwrap();
        let sse_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .header("accept", "text/event-stream")
            .body(Body::empty())
            .unwrap();
        let sse_resp = app.clone().oneshot(sse_req).await.unwrap();
        assert_eq!(sse_resp.status(), StatusCode::OK);
        let body = sse_resp.into_body();

        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::ToolListChanged,
        );
        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::ResourceListChanged,
        );

        let events = collect_sse_events(body, 3).await;
        let tool_event = events
            .iter()
            .find(|event| event.contains("notifications/tools/list_changed"))
            .expect("tool list changed event");
        let last_seen_id = sse_event_id(tool_event).expect("event id");

        let replay_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .header("accept", "text/event-stream")
            .header("last-event-id", last_seen_id.to_string())
            .body(Body::empty())
            .unwrap();
        let replay_resp = app.oneshot(replay_req).await.unwrap();
        assert_eq!(replay_resp.status(), StatusCode::OK);

        let replayed = collect_sse_events(replay_resp.into_body(), 2).await;
        assert!(
            replayed
                .iter()
                .any(|event| event.contains("notifications/resources/list_changed")),
            "expected replayed resource list_changed after id {last_seen_id}, got {replayed:?}"
        );
    }

    #[tokio::test]
    async fn targeted_progress_reaches_http_sse_session() {
        let state = test_state();
        let app = build_router(state.clone());

        let init_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let init_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&init_body).unwrap()))
            .unwrap();

        let init_resp = app.clone().oneshot(init_req).await.unwrap();
        let session_id = init_resp
            .headers()
            .get(SESSION_ID_HEADER)
            .and_then(|value| value.to_str().ok())
            .expect("session id header")
            .to_string();

        let sse_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .header("accept", "text/event-stream")
            .body(Body::empty())
            .unwrap();

        let sse_resp = app.oneshot(sse_req).await.unwrap();
        assert_eq!(sse_resp.status(), StatusCode::OK);
        let body = sse_resp.into_body();

        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::Progress {
                target: crate::notifications::NotificationTarget::Http {
                    session_id: Arc::from(session_id),
                },
                params: ProgressNotificationParam::new(
                    ProgressToken(NumberOrString::String(Arc::from("http-progress"))),
                    0.5,
                )
                .with_message("halfway"),
            },
        );

        let events = collect_sse_events(body, 3).await;
        assert!(
            events.iter().any(|event| {
                event.contains("notifications/progress")
                    && event.contains("http-progress")
                    && event.contains("halfway")
            }),
            "expected SSE stream to contain targeted progress notification, got {events:?}"
        );
    }

    #[tokio::test]
    async fn auth_state_changed_reaches_http_sse_as_logging_message() {
        let state = test_state();
        let app = build_router(state.clone());

        let init_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let init_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&init_body).unwrap()))
            .unwrap();

        let init_resp = app.clone().oneshot(init_req).await.unwrap();
        let session_id = init_resp
            .headers()
            .get(SESSION_ID_HEADER)
            .and_then(|value| value.to_str().ok())
            .expect("session id header")
            .to_string();

        let sse_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .header("accept", "text/event-stream")
            .body(Body::empty())
            .unwrap();

        let sse_resp = app.oneshot(sse_req).await.unwrap();
        assert_eq!(sse_resp.status(), StatusCode::OK);
        let body = sse_resp.into_body();

        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::AuthStateChanged {
                server_id: Arc::from("github"),
                new_state: crate::types::ServerHealth::AuthRequired,
            },
        );

        let events = collect_sse_events(body, 3).await;
        assert!(
            events.iter().any(|event| {
                event.contains("notifications/message")
                    && event.contains("auth_state_changed")
                    && event.contains("github")
                    && event.contains("AuthRequired")
            }),
            "expected SSE stream to contain auth state logging notification, got {events:?}"
        );
    }

    #[tokio::test]
    async fn token_refresh_exchanged_reaches_http_sse_as_logging_message() {
        let state = test_state();
        let app = build_router(state.clone());

        let init_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {
                    "name": "test-client",
                    "version": "1.0"
                }
            }
        });

        let init_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("content-type", "application/json")
            .body(Body::from(serde_json::to_vec(&init_body).unwrap()))
            .unwrap();

        let init_resp = app.clone().oneshot(init_req).await.unwrap();
        let session_id = init_resp
            .headers()
            .get(SESSION_ID_HEADER)
            .and_then(|value| value.to_str().ok())
            .expect("session id header")
            .to_string();

        let sse_req = HttpRequest::builder()
            .method("GET")
            .uri("/mcp")
            .header(SESSION_ID_HEADER, &session_id)
            .header("accept", "text/event-stream")
            .body(Body::empty())
            .unwrap();

        let sse_resp = app.oneshot(sse_req).await.unwrap();
        assert_eq!(sse_resp.status(), StatusCode::OK);
        let body = sse_resp.into_body();

        state.router.publish_protocol_notification(
            crate::notifications::ProtocolNotification::TokenRefreshExchanged {
                server_id: Arc::from("github"),
            },
        );

        let events = collect_sse_events(body, 3).await;
        assert!(
            events.iter().any(|event| {
                event.contains("notifications/message")
                    && event.contains("token_refresh_exchanged")
                    && event.contains("github")
            }),
            "expected SSE stream to contain token refresh logging notification, got {events:?}"
        );
    }

    // -- Bearer auth middleware tests --

    fn test_state_with_auth(token: &str) -> Arc<HttpState> {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Bearer,
            downstream_oauth: None,
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: Some(Arc::from(token)),
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        })
    }

    #[tokio::test]
    async fn auth_required_no_header_returns_401() {
        let state = test_state_with_auth("test_token_abc123");
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Content-Type", "application/json")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(resp.headers().get("WWW-Authenticate").unwrap(), "Bearer");
    }

    #[tokio::test]
    async fn auth_required_invalid_token_returns_401() {
        let state = test_state_with_auth("correct_token");
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Content-Type", "application/json")
            .header("Authorization", "Bearer wrong_token")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_required_valid_token_passes_through() {
        let token = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
        let state = test_state_with_auth(token);
        let app = build_router(state);

        // Valid token should pass auth and reach the handler (which will fail
        // on content type, not on auth — proving auth middleware passed)
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Authorization", format!("Bearer {token}"))
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        // Should get past auth (not 401) — will hit content type check (415)
        assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
    }

    #[tokio::test]
    async fn no_auth_required_loopback_passes_through() {
        // State with auth_token = None (loopback)
        let state = test_state();
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        // Should NOT be 401 — should hit content type check instead
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_valid_token_bypasses_origin_check() {
        let token = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
        let state = test_state_with_auth(token);
        let app = build_router(state);

        // Remote origin with valid bearer token — should bypass origin check
        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Authorization", format!("Bearer {token}"))
            .header("Origin", "https://remote-client.example.com")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        // Should NOT be 403 (origin rejected) — should pass through to handler
        assert_ne!(resp.status(), StatusCode::FORBIDDEN);
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn oauth_mode_metadata_endpoint_returns_server_metadata() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: None,
                    oauth_scopes: vec!["tools:read".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/oauth-authorization-server")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .expect("body bytes");
        let value: serde_json::Value = serde_json::from_slice(&body).expect("valid json");
        assert_eq!(
            value["authorization_endpoint"],
            "https://plug.example.com/oauth/authorize"
        );
        assert_eq!(
            value["token_endpoint"],
            "https://plug.example.com/oauth/token"
        );
        assert_eq!(value["scopes_supported"][0], "tools:read");
        assert_eq!(
            value["token_endpoint_auth_methods_supported"],
            json!(["none"])
        );
    }

    #[tokio::test]
    async fn oauth_mode_metadata_endpoint_limits_auth_methods_for_confidential_client() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: Some("secret-123".to_string().into()),
                    oauth_scopes: vec!["tools:read".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/oauth-authorization-server")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .expect("body bytes");
        let value: serde_json::Value = serde_json::from_slice(&body).expect("valid json");
        assert_eq!(
            value["token_endpoint_auth_methods_supported"],
            json!(["client_secret_basic", "client_secret_post"])
        );
        assert_eq!(
            value["grant_types_supported"],
            json!(["authorization_code", "refresh_token", "client_credentials"])
        );
    }

    #[tokio::test]
    async fn oauth_protected_resource_metadata_endpoint_returns_rfc9728_metadata() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: Some("secret-123".to_string().into()),
                    oauth_scopes: vec!["tools:read".to_string(), "offline_access".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/oauth-protected-resource")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .expect("body bytes");
        let value: serde_json::Value = serde_json::from_slice(&body).expect("valid json");
        assert_eq!(value["resource"], "https://plug.example.com/mcp");
        assert_eq!(
            value["authorization_servers"],
            json!(["https://plug.example.com"])
        );
        assert_eq!(value["scopes_supported"], json!(["tools:read"]));
        assert_eq!(value["bearer_methods_supported"], json!(["header"]));
    }

    #[tokio::test]
    async fn oauth_client_credentials_flow_issues_bearer_without_refresh_token() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: Some("secret-123".to_string().into()),
                    oauth_scopes: vec!["tools:read".to_string(), "offline_access".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let token_req = HttpRequest::builder()
            .method("POST")
            .uri("/oauth/token")
            .header("Content-Type", "application/x-www-form-urlencoded")
            .header(header::AUTHORIZATION, "Basic Y2xpZW50LTEyMzpzZWNyZXQtMTIz")
            .body(Body::from(
                "grant_type=client_credentials&scope=tools%3Aread+offline_access",
            ))
            .unwrap();

        let token_resp = app.clone().oneshot(token_req).await.unwrap();
        assert_eq!(token_resp.status(), StatusCode::OK);
        let body = axum::body::to_bytes(token_resp.into_body(), 10_000)
            .await
            .expect("body bytes");
        let value: serde_json::Value = serde_json::from_slice(&body).expect("valid json");
        assert!(value["access_token"].as_str().is_some());
        assert!(value.get("refresh_token").is_none());
        assert_eq!(value["scope"], "tools:read");

        let mcp_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header(
                header::AUTHORIZATION,
                format!("Bearer {}", value["access_token"].as_str().unwrap()),
            )
            .body(Body::from("{}"))
            .unwrap();
        let mcp_resp = app.oneshot(mcp_req).await.unwrap();
        assert_eq!(mcp_resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
    }

    #[tokio::test]
    async fn oauth_mode_requires_access_token_for_mcp_requests() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: None,
                    oauth_scopes: vec!["tools:read".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Content-Type", "application/json")
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(
            resp.headers()
                .get(header::WWW_AUTHENTICATE)
                .and_then(|v| v.to_str().ok())
                .expect("www-authenticate"),
            "Bearer resource_metadata=\"https://plug.example.com/.well-known/oauth-protected-resource\", scope=\"tools:read\""
        );
    }

    #[tokio::test]
    async fn oauth_authorization_code_flow_issues_working_access_token() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: Some("secret-123".to_string().into()),
                    oauth_scopes: vec!["tools:read".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let authorize_req = HttpRequest::builder()
            .method("GET")
            .uri("/oauth/authorize?response_type=code&client_id=client-123&redirect_uri=https%3A%2F%2Fclient.example.com%2Fcallback&state=abc123&code_challenge=E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM&code_challenge_method=S256&scope=tools:read")
            .body(Body::empty())
            .unwrap();
        let authorize_resp = app.clone().oneshot(authorize_req).await.unwrap();
        assert_eq!(authorize_resp.status(), StatusCode::FOUND);
        let location = authorize_resp
            .headers()
            .get(header::LOCATION)
            .and_then(|v| v.to_str().ok())
            .expect("redirect location");
        let code = location
            .split("code=")
            .nth(1)
            .and_then(|v| v.split('&').next())
            .expect("authorization code");

        let token_req = HttpRequest::builder()
            .method("POST")
            .uri("/oauth/token")
            .header("Content-Type", "application/x-www-form-urlencoded")
            .header(
                header::AUTHORIZATION,
                "Basic Y2xpZW50LTEyMzpzZWNyZXQtMTIz",
            )
            .body(Body::from(format!(
                "grant_type=authorization_code&code={code}&redirect_uri=https%3A%2F%2Fclient.example.com%2Fcallback&code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
            )))
            .unwrap();
        let token_resp = app.clone().oneshot(token_req).await.unwrap();
        assert_eq!(token_resp.status(), StatusCode::OK);
        let token_body = axum::body::to_bytes(token_resp.into_body(), 10_000)
            .await
            .expect("token body");
        let token_value: serde_json::Value =
            serde_json::from_slice(&token_body).expect("token json");
        let access_token = token_value["access_token"]
            .as_str()
            .expect("access token")
            .to_string();
        let refresh_token = token_value["refresh_token"]
            .as_str()
            .expect("refresh token")
            .to_string();
        assert_eq!(token_value["token_type"], "Bearer");

        let mcp_req = HttpRequest::builder()
            .method("POST")
            .uri("/mcp")
            .header("Content-Type", "application/json")
            .header("Authorization", format!("Bearer {access_token}"))
            .body(Body::from("{}"))
            .unwrap();
        let mcp_resp = app.clone().oneshot(mcp_req).await.unwrap();
        assert_eq!(mcp_resp.status(), StatusCode::BAD_REQUEST);

        let refresh_req = HttpRequest::builder()
            .method("POST")
            .uri("/oauth/token")
            .header("Content-Type", "application/x-www-form-urlencoded")
            .header(header::AUTHORIZATION, "Basic Y2xpZW50LTEyMzpzZWNyZXQtMTIz")
            .body(Body::from(format!(
                "grant_type=refresh_token&refresh_token={refresh_token}"
            )))
            .unwrap();
        let refresh_resp = app.oneshot(refresh_req).await.unwrap();
        assert_eq!(refresh_resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn discovery_server_card_when_unauth_on_protected_server_is_static() {
        let state = test_state_with_auth("secret_token");
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp.json")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .unwrap();
        let card: serde_json::Value = serde_json::from_slice(&body).unwrap();

        assert!(card.get("servers").is_none());
        assert!(card.get("tools").is_none());
        assert_eq!(card["remotes"][0]["url"], "/mcp");
        assert_eq!(card["remotes"][0]["headers"][0]["name"], "Authorization");
        assert_eq!(card["remotes"][0]["headers"][0]["isSecret"], true);
    }

    #[tokio::test]
    async fn discovery_server_card_when_unauth_on_oauth_protected_server_is_static() {
        let sm = Arc::new(crate::server::ServerManager::new());
        let router = Arc::new(ToolRouter::new(
            sm,
            crate::proxy::RouterConfig {
                prefix_delimiter: "__".to_string(),
                priority_tools: Vec::new(),
                disabled_tools: Vec::new(),
                tool_description_max_chars: None,
                tool_search_threshold: 50,
                meta_tool_mode: false,
                lazy_tools: crate::config::LazyToolsConfig::default(),
                tool_filter_enabled: true,
                enrichment_servers: std::collections::HashSet::new(),
            },
        ));
        let state = Arc::new(HttpState {
            router,
            sessions: Arc::new(crate::session::StatefulSessionStore::new(1800, 100)),
            cancel: CancellationToken::new(),
            auth_mode: crate::config::DownstreamAuthMode::Oauth,
            downstream_oauth: Some(crate::downstream_oauth::DownstreamOauthManager::new(
                crate::downstream_oauth::DownstreamOauthConfig {
                    public_base_url: "https://plug.example.com".to_string(),
                    oauth_client_id: Some("client-123".to_string()),
                    oauth_client_secret: None,
                    oauth_scopes: vec!["tools:read".to_string()],
                },
            )),
            sse_channel_capacity: 32,
            allowed_origins: Vec::new(),
            notification_task_started: AtomicBool::new(false),
            auth_token: None,
            roots_capable_sessions: DashMap::new(),
            pending_client_requests: DashMap::new(),
            reverse_request_counter: AtomicU64::new(1),
            client_capabilities: DashMap::new(),
        });
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp.json")
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .unwrap();
        let card: serde_json::Value = serde_json::from_slice(&body).unwrap();

        assert!(card.get("servers").is_none());
        assert!(card.get("tools").is_none());
        assert_eq!(card["remotes"][0]["headers"][0]["name"], "Authorization");
    }

    #[tokio::test]
    async fn discovery_server_card_does_not_expand_when_authenticated() {
        let token = "secret_token";
        let state = test_state_with_auth(token);
        let app = build_router(state);

        let req = HttpRequest::builder()
            .method("GET")
            .uri("/.well-known/mcp.json")
            .header("Authorization", format!("Bearer {token}"))
            .body(Body::empty())
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), 10_000)
            .await
            .unwrap();
        let card: serde_json::Value = serde_json::from_slice(&body).unwrap();

        assert!(card.get("servers").is_none());
        assert!(card.get("tools").is_none());
        assert_eq!(card["remotes"][0]["headers"][0]["name"], "Authorization");
    }
}
