mcpls_core/

transport.rs

//! Transport selection for the MCP server.
//!
//! This module defines the [`Transport`] enum that controls how the MCP server
//! communicates with clients. Stdio is always available; HTTP transport is
//! opt-in via the `transport-http` Cargo feature.
//!
//! # Selecting a transport
//!
//! Pass a [`Transport`] value to [`crate::serve_with`] to choose the runtime
//! binding. The default entry point [`crate::serve`] always uses
//! [`Transport::Stdio`].

/// The transport over which the MCP server communicates with clients.
///
/// # Examples
///
/// See [`crate::serve_with`]'s "Shutdown" section before copying this
/// verbatim: under [`Transport::Stdio`], `main` must call
/// `std::process::exit` rather than returning normally, or `SIGTERM`/
/// `SIGINT` can hang while an MCP client's stdin write end is still open
/// (#308).
///
/// ```rust,ignore
/// use mcpls_core::{Transport, serve_with, ServerConfig};
///
/// #[tokio::main]
/// async fn main() {
///     let config = ServerConfig::load().expect("failed to load config");
///     let result = serve_with(config, Transport::Stdio).await;
///     std::process::exit(if result.is_ok() { 0 } else { 1 });
/// }
/// ```
#[non_exhaustive]
pub enum Transport {
    /// Standard I/O transport (default).
    ///
    /// Reads from `stdin` and writes to `stdout`. This is the transport used
    /// by MCP clients that launch mcpls as a child process.
    Stdio,

    /// Streamable HTTP transport (MCP spec 2025-11-25).
    ///
    /// Binds a TCP listener and serves the MCP protocol over HTTP, enabling
    /// network-accessible deployments and clients that speak HTTP rather than
    /// stdio. Only available when the `transport-http` feature is enabled.
    #[cfg(feature = "transport-http")]
    #[cfg_attr(docsrs, doc(cfg(feature = "transport-http")))]
    Http(HttpConfig),
}

/// Configuration for the HTTP transport.
///
/// Passed inside [`Transport::Http`] to control the TCP bind address and the
/// URL path the MCP service is mounted at.
///
/// # Note on DNS rebinding
///
/// `rmcp`'s `StreamableHttpService` validates the `Host` header against an
/// allow-list that defaults to loopback addresses only (`localhost`,
/// `127.0.0.1`, `::1`). If you bind to `0.0.0.0` or a non-loopback address,
/// clients must send requests with a `Host` that matches the allow-list, or
/// use a reverse proxy that rewrites the `Host` header.
///
/// # Examples
///
/// ```rust,ignore
/// use std::net::SocketAddr;
/// use mcpls_core::{HttpConfig, Transport};
///
/// let cfg = HttpConfig::new("127.0.0.1:3000".parse().unwrap(), "/mcp");
/// let transport = Transport::Http(cfg);
/// ```
#[cfg(feature = "transport-http")]
#[cfg_attr(docsrs, doc(cfg(feature = "transport-http")))]
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct HttpConfig {
    /// TCP address to bind (e.g. `127.0.0.1:3000`).
    pub bind: std::net::SocketAddr,
    /// URL path prefix the MCP service is mounted at (e.g. `"/mcp"`).
    pub path: String,
    /// Maximum size, in bytes, of a single POST request body.
    ///
    /// Enforced by `rmcp`'s `StreamableHttpService` while streaming the body,
    /// independent of `Content-Length` or chunked transfer encoding. Requests
    /// exceeding this limit receive `413 Payload Too Large`. Defaults to
    /// [`HttpConfig::DEFAULT_MAX_REQUEST_BODY_BYTES`] (4 MiB), which
    /// comfortably covers MCP tool-call request bodies (large results, e.g.
    /// from `workspace/symbol` or bulk edits, are returned in the response,
    /// which this limit does not constrain). A value of `0` rejects every
    /// POST body.
    pub max_request_body_bytes: usize,
    /// Maximum number of concurrent HTTP sessions.
    ///
    /// This is a hard bound, enforced atomically at session creation via a
    /// semaphore — never more than this many sessions can be active at once,
    /// regardless of request concurrency.
    /// Requests that would start a new session beyond this limit receive
    /// `429 Too Many Requests`. Defaults to
    /// [`HttpConfig::DEFAULT_MAX_CONCURRENT_SESSIONS`]. A value of `0`
    /// rejects every session.
    pub max_concurrent_sessions: usize,
}

#[cfg(feature = "transport-http")]
#[cfg_attr(docsrs, doc(cfg(feature = "transport-http")))]
impl HttpConfig {
    /// Default request body size cap (4 MiB), matching `rmcp`'s own default.
    pub const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 4 * 1024 * 1024;
    /// Default concurrent HTTP session cap.
    pub const DEFAULT_MAX_CONCURRENT_SESSIONS: usize = 100;

    /// Create an [`HttpConfig`] with default body-size and session caps.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// use mcpls_core::HttpConfig;
    ///
    /// let cfg = HttpConfig::new("127.0.0.1:3000".parse().unwrap(), "/mcp");
    /// ```
    pub fn new(bind: std::net::SocketAddr, path: impl Into<String>) -> Self {
        Self {
            bind,
            path: path.into(),
            max_request_body_bytes: Self::DEFAULT_MAX_REQUEST_BODY_BYTES,
            max_concurrent_sessions: Self::DEFAULT_MAX_CONCURRENT_SESSIONS,
        }
    }

    /// Override the maximum POST request body size in bytes.
    #[must_use]
    pub const fn with_max_request_body_bytes(mut self, bytes: usize) -> Self {
        self.max_request_body_bytes = bytes;
        self
    }

    /// Override the maximum number of concurrent HTTP sessions.
    #[must_use]
    pub const fn with_max_concurrent_sessions(mut self, max: usize) -> Self {
        self.max_concurrent_sessions = max;
        self
    }
}

use rmcp::ServiceExt as _;
#[cfg(feature = "transport-http")]
use rmcp::model::{ClientJsonRpcMessage, ServerJsonRpcMessage};
#[cfg(feature = "transport-http")]
use rmcp::transport::streamable_http_server::session::local::{
    LocalSessionManager, LocalSessionManagerError,
};
#[cfg(feature = "transport-http")]
use rmcp::transport::streamable_http_server::session::{
    ServerSseMessage, SessionId, SessionManager,
};

/// A registered handle for waiting on a shutdown signal: `SIGTERM`/`SIGINT`
/// on Unix (as sent by containers, systemd, and `Ctrl-C`) or `Ctrl-C` on
/// Windows.
///
/// Constructed once by [`crate::serve_with`], *before* any startup work
/// (LSP-server discovery heuristics, `spawn_lsp_servers_background`) runs,
/// and moved by value into whichever transport (`run_stdio`/`run_http`) ends
/// up serving. Registering this early — rather than inside the transport
/// function itself — closes the startup window between process start and the
/// transport loop, during which a signal would otherwise hit the OS's
/// default disposition (immediate termination, bypassing
/// [`crate::bridge::Translator::shutdown_servers`] and risking an orphaned
/// LSP child process that `spawn_lsp_servers_background` is mid-spawning;
/// see #270).
///
/// Every signal kind is held as its own persistent stream
/// (`tokio::signal::unix::Signal` / `tokio::signal::windows::CtrlC`) for the
/// lifetime of this value, rather than re-registered on every
/// [`ShutdownSignal::recv`] call via `tokio::signal::ctrl_c()`: a signal
/// delivered while a *specific* listener isn't being polled is only observed
/// by that same listener's next poll — a freshly (re-)subscribed one starts
/// at the broadcast's current version and never sees it (tokio
/// `signal/registry.rs`). Since [`recv`](ShutdownSignal::recv) is awaited
/// from more than one call site — both by [`run_stdio`], which races it
/// against the MCP handshake and then the post-handshake serve loop, and
/// across the gap between construction in `serve_with` and the first await
/// inside the transport — a fresh registration per call would risk losing a
/// signal delivered in between.
///
/// This instance is dropped as soon as the transport function it was moved
/// into returns — but that does *not* deregister the OS-level handler:
/// `tokio::signal` installs it once per process and never uninstalls it, no
/// matter how many `ShutdownSignal`s are constructed or dropped. What
/// dropping the last live instance actually does is remove the only
/// receiver a delivered signal could be broadcast to, so until a new one
/// subscribes, a signal is recorded and then silently discarded rather than
/// observed by anything — making that stretch of code uninterruptible
/// rather than unsafe. [`crate::shutdown`] (the post-transport cleanup run
/// immediately after) registers a *second* `ShutdownSignal` of its own so a
/// repeat signal during cleanup has a receiver again and can force an exit;
/// see #329.
pub(crate) struct ShutdownSignal {
    #[cfg(unix)]
    sigterm: Option<tokio::signal::unix::Signal>,
    #[cfg(unix)]
    sigint: Option<tokio::signal::unix::Signal>,
    #[cfg(windows)]
    ctrl_c: Option<tokio::signal::windows::CtrlC>,
}

impl ShutdownSignal {
    /// Registers the process's shutdown signal handler(s) up front.
    pub(crate) fn new() -> Self {
        #[cfg(unix)]
        {
            use tokio::signal::unix::{SignalKind, signal};
            let sigterm = match signal(SignalKind::terminate()) {
                Ok(sigterm) => Some(sigterm),
                Err(e) => {
                    tracing::warn!(
                        "SIGTERM handler registration failed ({e}), SIGTERM will not be caught"
                    );
                    None
                }
            };
            let sigint = match signal(SignalKind::interrupt()) {
                Ok(sigint) => Some(sigint),
                Err(e) => {
                    tracing::warn!(
                        "SIGINT handler registration failed ({e}), SIGINT will not be caught"
                    );
                    None
                }
            };
            Self { sigterm, sigint }
        }
        #[cfg(windows)]
        {
            let ctrl_c = match tokio::signal::windows::ctrl_c() {
                Ok(ctrl_c) => Some(ctrl_c),
                Err(e) => {
                    tracing::warn!("Ctrl-C handler registration failed ({e})");
                    None
                }
            };
            Self { ctrl_c }
        }
        #[cfg(not(any(unix, windows)))]
        {
            Self {}
        }
    }

    /// Waits for the next shutdown signal. May be awaited repeatedly.
    pub(crate) async fn recv(&mut self) {
        #[cfg(unix)]
        {
            match (self.sigterm.as_mut(), self.sigint.as_mut()) {
                (Some(sigterm), Some(sigint)) => {
                    tokio::select! {
                        _ = sigterm.recv() => {},
                        _ = sigint.recv() => {},
                    }
                }
                (Some(sigterm), None) => {
                    sigterm.recv().await;
                }
                (None, Some(sigint)) => {
                    sigint.recv().await;
                }
                (None, None) => {
                    // Both registrations failed above; fall back to a
                    // one-shot listener so shutdown is still possible, even
                    // though it doesn't carry the same across-calls
                    // durability the held streams above do (see the struct
                    // docs).
                    let _ = tokio::signal::ctrl_c().await;
                }
            }
        }
        #[cfg(windows)]
        {
            match self.ctrl_c.as_mut() {
                Some(ctrl_c) => {
                    ctrl_c.recv().await;
                }
                None => {
                    let _ = tokio::signal::ctrl_c().await;
                }
            }
        }
        #[cfg(not(any(unix, windows)))]
        {
            // No persistent listener is available on this platform; same
            // caveat as the Unix double-registration-failure fallback above.
            let _ = tokio::signal::ctrl_c().await;
        }
    }
}

/// Run the MCP server over stdio.
///
/// Serves the given `mcp_server` using stdin/stdout and populates `peer_cell`
/// once the transport is established so that diagnostic pump tasks can begin
/// forwarding `resources/updated` notifications. Returns as soon as either
/// the stdio transport closes (client disconnect / stdin EOF) or a `SIGTERM`/
/// `SIGINT` is received, so callers can run orderly cleanup — such as
/// [`crate::bridge::Translator::shutdown_servers`] — before the process
/// exits. `shutdown_signal` is dropped when this function returns, and this
/// function does no draining of its own after a signal arrives — so a
/// repeat signal during the post-return cleanup in [`crate::shutdown`] is
/// caught only by the second `ShutdownSignal` that function registers for
/// itself, not by this one (see [`ShutdownSignal`]'s docs and #329).
///
/// `shutdown_signal` is constructed by [`crate::serve_with`] *before* any
/// startup work runs (see [`ShutdownSignal`]'s docs) and is raced here
/// against both the MCP handshake and, once it completes, the
/// post-handshake serve loop. `serve(..)` awaits the full MCP `initialize`
/// handshake internally (reading the client's request and writing the
/// response) before resolving, so a signal arriving during that wait — which
/// can be indefinite if the client is slow to send `initialize` — must be
/// caught there too, not only after the handshake finishes. On signal, the
/// in-flight handshake or `RunningService` is dropped rather than awaited to
/// completion; `rmcp` closes it asynchronously in that case, which is
/// acceptable here since the process exits shortly after -- callers must
/// exit via `std::process::exit` rather than returning normally from `main`,
/// or an uncancellable `tokio::io::stdin()` blocking thread can stall
/// runtime shutdown indefinitely (see `mcpls-cli`'s `main.rs` and #308).
pub(crate) async fn run_stdio(
    mcp_server: crate::mcp::McplsServer,
    peer_cell: &tokio::sync::OnceCell<rmcp::Peer<rmcp::RoleServer>>,
    mut shutdown_signal: ShutdownSignal,
) -> Result<(), crate::Error> {
    let service = tokio::select! {
        result = mcp_server.serve(rmcp::transport::stdio()) => {
            result.map_err(|e| crate::Error::McpServer(format!("Failed to start MCP server: {e}")))?
        }
        () = shutdown_signal.recv() => {
            tracing::info!("shutdown signal received during handshake, stopping stdio transport");
            return Ok(());
        }
    };

    if let Err(e) = peer_cell.set(service.peer().clone()) {
        tracing::debug!("Peer cell already set ({}), ignoring", e);
    }

    tokio::select! {
        result = service.waiting() => result
            .map(|_| ())
            .map_err(|e| crate::Error::McpServer(format!("MCP server error: {e}"))),
        () = shutdown_signal.recv() => {
            tracing::info!("shutdown signal received, stopping stdio transport");
            Ok(())
        }
    }
}

/// Run the MCP server over Streamable HTTP (MCP spec 2025-11-25).
///
/// Binds `cfg.bind`, mounts the MCP service at `cfg.path` (and `/`), and
/// serves until `Ctrl-C` or `SIGTERM` is received.
///
/// Each HTTP session receives its own `McplsServer` instance (see
/// [`crate::mcp::McplsServer::for_new_session`]). The shared `Arc<Translator>`
/// inside is the same across all sessions, so LSP state is still global per
/// process.
///
/// # Note
///
/// Diagnostic push notifications (`resources/updated`) are not forwarded to
/// HTTP sessions in this release — the single-peer pump architecture from
/// stdio is kept as-is. Clients can still poll diagnostics via the existing
/// MCP tools. A follow-up issue will add per-session broadcast.
///
/// On rmcp's stateless request path, "one instance per session" narrows to
/// "one instance per request"; `resources/subscribe`/`unsubscribe` detect
/// that path and return an explicit error rather than silently accepting a
/// subscription that would never be observed -- see
/// [`SubscriptionRegistry`](crate::bridge::SubscriptionRegistry)'s "Known
/// limitation" section.
///
/// # Resource limits
///
/// POST bodies exceeding `cfg.max_request_body_bytes` are rejected with
/// `413 Payload Too Large` (enforced by `rmcp`). Once `cfg.max_concurrent_sessions`
/// sessions are active, a request that would start a new one is rejected with
/// `429 Too Many Requests` — enforced as a hard bound at session creation by
/// [`CappedSessionManager`] and surfaced over HTTP by [`enforce_session_cap`].
///
/// # Shutdown
///
/// On `SIGTERM`/`SIGINT`, in-flight connections get up to
/// [`HTTP_GRACEFUL_SHUTDOWN_TIMEOUT`] to finish before this function returns
/// regardless — bounding shutdown this way lets the caller run its own
/// post-shutdown cleanup (e.g. closing registered LSP servers) even if a
/// connection never observes the cancellation (a stuck SSE stream, say).
/// `shutdown_signal` is constructed by [`crate::serve_with`] before any
/// startup work runs (see [`ShutdownSignal`]'s docs), so its registration
/// predates this function's own `TcpListener::bind` call — a signal between
/// bind and the graceful-shutdown future's first poll is still caught.
/// `shutdown_signal` is moved into (and dropped by) the
/// `with_graceful_shutdown` closure below once it resolves — i.e. as soon as
/// the *first* signal is received, well before this function returns. A
/// second, freshly constructed `ShutdownSignal` then covers the
/// connection-drain wait that follows (bounded by
/// [`HTTP_GRACEFUL_SHUTDOWN_TIMEOUT`]): a repeat signal caught there cuts the
/// drain short (dropping `serve` the same way the timeout branch already
/// does) instead of making the operator wait out the full timeout.
///
/// Cutting the drain short is *not* an immediate process exit: this function
/// still returns `Ok(())` normally, and its caller ([`crate::serve_with`])
/// proceeds straight into the ordinary post-transport shutdown sequence
/// ([`crate::shutdown`] — LSP server shutdown plus any pending background
/// init task, bounded by its own ~15s worst case). [`crate::shutdown`]'s own
/// registration (#329) takes over once *this* function returns, covering
/// that cleanup window and escalating to a forced `std::process::exit(1)` on
/// any *further* repeat signal — so an operator wanting a true immediate exit
/// needs a third signal, not a second. This is a deliberate choice, not an
/// oversight: calling `exit(1)` directly from this branch would skip
/// unwinding and forfeit `kill_on_drop` cleanup of any still-running LSP
/// child processes, which is worse than requiring one more signal.
#[cfg(feature = "transport-http")]
// `session_manager` and `service` are moved into `app`, which is served until
// shutdown — clippy's drop-tightening heuristic misreads that as an
// early-droppable temporary because both types embed `tokio::sync` lock types
// (`CappedSessionManager`'s `Mutex`, `StreamableHttpService`'s `RwLock`s).
#[allow(clippy::significant_drop_tightening)]
pub(crate) async fn run_http(
    mcp_server: crate::mcp::McplsServer,
    cfg: HttpConfig,
    mut shutdown_signal: ShutdownSignal,
) -> Result<(), crate::Error> {
    use std::sync::Arc;

    use rmcp::transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService,
    };
    use tokio_util::sync::CancellationToken;

    let session_manager = Arc::new(CappedSessionManager::new(cfg.max_concurrent_sessions));
    let cancel = CancellationToken::new();

    // `mcp_server` is moved (not cloned): `McplsServer` is deliberately not
    // `Clone` (#478) so this is the only value the factory below can build
    // new sessions from, rather than a shared instance a caller could
    // accidentally hand to multiple sessions.
    let mcp_for_factory = mcp_server;
    // StreamableHttpServerConfig is #[non_exhaustive]; construct via Default then mutate.
    let mut http_cfg = StreamableHttpServerConfig::default();
    http_cfg.cancellation_token = cancel.clone();
    http_cfg.max_request_body_bytes = cfg.max_request_body_bytes;

    // `for_new_session`, not `.clone()`: every session must get its own
    // `ResourceSubscriptions` set (#478) rather than sharing `mcp_for_factory`'s,
    // while still sharing its `Arc<Translator>` and the rest of the LSP-facing
    // state via a cheap `Arc` bump. On rmcp's stateless path this factory runs
    // once per request, not per session -- see `SubscriptionRegistry`'s
    // "Known limitation" doc.
    let service = StreamableHttpService::new(
        move || Ok::<_, std::io::Error>(mcp_for_factory.for_new_session()),
        session_manager,
        http_cfg,
    );

    let app = axum::Router::new()
        .nest_service(&cfg.path, service.clone())
        .route_service("/", service)
        .layer(axum::middleware::from_fn(enforce_session_cap));

    let listener = tokio::net::TcpListener::bind(cfg.bind)
        .await
        .map_err(|e| crate::Error::McpServer(format!("bind {}: {e}", cfg.bind)))?;

    tracing::info!(addr = %cfg.bind, path = %cfg.path, "MCP HTTP transport listening");
    if !cfg.bind.ip().is_loopback() {
        tracing::warn!(
            addr = %cfg.bind,
            "binding to a non-loopback address: mcpls performs no authentication of its own on \
             any transport — place this endpoint behind a reverse proxy that enforces \
             authentication. The proxy must also rewrite the Host header, since rmcp's Host \
             validation allows only localhost/127.0.0.1/::1 by default"
        );
    }

    // `cancel` is cancelled exactly once, when the shutdown signal fires
    // (below). Cloned first so the force-timeout and repeat-signal branches
    // can each observe that same moment independently of the
    // `with_graceful_shutdown` closure, which consumes its own clone.
    let cancel_for_force_timeout = cancel.clone();
    let cancel_for_repeat_signal = cancel.clone();
    let serve = axum::serve(listener, app).with_graceful_shutdown(async move {
        shutdown_signal.recv().await;
        cancel.cancel();
    });

    // The force-timeout only starts counting once `cancel` is actually
    // cancelled — i.e. once a shutdown signal has been received — not from
    // server startup. Without that ordering, `tokio::time::timeout` wrapping
    // `serve` directly would tear down the listener after
    // `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` of ordinary uptime, signal or not.
    // This bounds only the "drain in-flight connections after shutdown was
    // requested" phase, so a connection that never observes `cancel` (e.g. a
    // stuck SSE stream) can't hang the caller's post-shutdown cleanup
    // (draining/closing LSP servers) indefinitely.
    tokio::select! {
        result = serve => result.map_err(|e| crate::Error::McpServer(format!("http serve: {e}"))),
        () = async move {
            cancel_for_force_timeout.cancelled().await;
            tokio::time::sleep(HTTP_GRACEFUL_SHUTDOWN_TIMEOUT).await;
        } => {
            tracing::warn!(
                timeout = ?HTTP_GRACEFUL_SHUTDOWN_TIMEOUT,
                "HTTP graceful shutdown did not complete in time, proceeding with shutdown anyway"
            );
            Ok(())
        }
        // #349: `shutdown_signal` above is consumed and dropped as soon as
        // the first signal arrives, leaving no listener for a repeat signal
        // during the connection-drain wait that follows. Wait for `cancel`
        // first and only then construct a fresh `ShutdownSignal` -- rather
        // than registering one up front, alongside `shutdown_signal` -- so
        // it starts with no pending signal of its own: `tokio::signal`
        // fans a delivered signal out to every live listener, so a listener
        // already registered before the first signal arrived would
        // independently observe that same signal and misreport it as a
        // repeat. This leaves a much smaller, accepted gap instead: between
        // `cancel.cancel()` firing and `ShutdownSignal::new()` actually
        // registering on the next scheduler hop, there is no pollable
        // listener at all, so a signal delivered in that sub-millisecond
        // window is lost. Registering up front would only trade this for
        // the coalescing problem above -- it is not fully closable either
        // way, and the window is far below human reaction time to a second
        // keypress.
        () = async move {
            cancel_for_repeat_signal.cancelled().await;
            let mut repeat_signal = ShutdownSignal::new();
            repeat_signal.recv().await;
        } => {
            tracing::warn!(
                "repeat shutdown signal received during HTTP connection drain, cutting drain short"
            );
            Ok(())
        }
    }
}

/// Upper bound [`run_http`] waits, once shutdown has been signaled, for
/// `axum`'s graceful shutdown to finish draining in-flight connections
/// before giving up and returning anyway.
#[cfg(feature = "transport-http")]
const HTTP_GRACEFUL_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// Wraps [`LocalSessionManager`], bounding concurrent HTTP sessions to a
/// fixed capacity.
///
/// A [`tokio::sync::Semaphore`] permit is acquired atomically inside
/// [`create_session`](SessionManager::create_session) — before delegating to
/// the inner manager — and held for the session's lifetime, released in
/// [`close_session`](SessionManager::close_session). This makes the cap a
/// true hard bound: the check and the reservation happen as one step, so no
/// number of concurrent requests can observe spare capacity and all proceed
/// past it (a "check-then-create" race that a separate read of the session
/// count could not avoid).
///
/// Enforcement lives here, at the `SessionManager` layer, rather than in Axum
/// middleware sniffing request headers, because that is the only place
/// guaranteed to run exactly when — and only when — a session is actually
/// created. `rmcp` 3.2.0's `StreamableHttpService::handle_post` classifies
/// every `initialize` request as legacy and always calls `create_session`,
/// whatever protocol version it names — the handshake only exists in
/// revisions before `2026-07-28`, so a version named in its params never
/// routes it to the stateless path. Only *non*-`initialize` requests that
/// carry SEP-2575 per-request `_meta` (`io.modelcontextprotocol/protocolVersion`
/// = `2026-07-28` plus the required `clientCapabilities` key), and
/// `server/discover` requests, take the stateless discover-lifecycle path
/// that never calls `create_session`. A header-based middleware heuristic
/// can't tell these apart without duplicating `rmcp`'s internal protocol
/// classification, so it either 429s traffic that never consumed a session
/// slot, or — in an all-stateless deployment — never fires at all.
///
/// `restore_session` and `event_store` deliberately use
/// [`SessionManager`]'s trait defaults (`NotSupported` / `None`) instead of
/// delegating to `inner`: `HttpConfig` exposes no session-store knob, so
/// these are unreachable today, but delegating them would let a restored
/// session skip the semaphore entirely — a cap bypass. Leave them as
/// defaults; overriding them to delegate is not a bug fix.
#[cfg(feature = "transport-http")]
struct CappedSessionManager {
    inner: LocalSessionManager,
    semaphore: std::sync::Arc<tokio::sync::Semaphore>,
    permits:
        tokio::sync::Mutex<std::collections::HashMap<SessionId, tokio::sync::OwnedSemaphorePermit>>,
}

#[cfg(feature = "transport-http")]
impl CappedSessionManager {
    fn new(max_sessions: usize) -> Self {
        Self {
            inner: LocalSessionManager::default(),
            semaphore: std::sync::Arc::new(tokio::sync::Semaphore::new(max_sessions)),
            permits: tokio::sync::Mutex::new(std::collections::HashMap::new()),
        }
    }
}

/// Marker embedded in [`CappedSessionManagerError::CapReached`]'s rendered
/// message.
///
/// `rmcp`'s `StreamableHttpService` always maps `create_session` failures to
/// a generic `500 Internal Server Error` (`internal_error_response` in
/// `server_side_http.rs` is a fixed, non-configurable mapping — `rmcp` gives
/// callers no other hook). [`enforce_session_cap`] looks for this marker in
/// the response body to translate a capacity rejection into
/// `429 Too Many Requests` without misclassifying other `create_session`
/// failures as capacity issues.
#[cfg(feature = "transport-http")]
const SESSION_CAP_MARKER: &str = "mcpls-http-session-cap-reached";

/// Error type for [`CappedSessionManager`].
#[cfg(feature = "transport-http")]
#[derive(Debug, thiserror::Error)]
enum CappedSessionManagerError {
    /// The concurrent-session cap was already reached.
    #[error("{SESSION_CAP_MARKER}: maximum concurrent HTTP sessions already active")]
    CapReached,
    /// The wrapped [`LocalSessionManager`] failed.
    #[error(transparent)]
    Inner(#[from] LocalSessionManagerError),
}

#[cfg(feature = "transport-http")]
impl SessionManager for CappedSessionManager {
    type Error = CappedSessionManagerError;
    type Transport = <LocalSessionManager as SessionManager>::Transport;

    async fn create_session(&self) -> Result<(SessionId, Self::Transport), Self::Error> {
        let permit = self
            .semaphore
            .clone()
            .try_acquire_owned()
            .map_err(|_| CappedSessionManagerError::CapReached)?;
        let (id, transport) = self.inner.create_session().await?;
        self.permits.lock().await.insert(id.clone(), permit);
        Ok((id, transport))
    }

    async fn initialize_session(
        &self,
        id: &SessionId,
        message: ClientJsonRpcMessage,
    ) -> Result<ServerJsonRpcMessage, Self::Error> {
        Ok(self.inner.initialize_session(id, message).await?)
    }

    async fn has_session(&self, id: &SessionId) -> Result<bool, Self::Error> {
        Ok(self.inner.has_session(id).await?)
    }

    async fn close_session(&self, id: &SessionId) -> Result<(), Self::Error> {
        // Release the permit unconditionally, before propagating any error from
        // `inner.close_session`: on error the inner manager has already dropped
        // the session from its own table (see `LocalSessionManager::close_session`),
        // so skipping the removal here would leak the permit permanently and
        // monotonically shrink capacity.
        self.permits.lock().await.remove(id);
        self.inner.close_session(id).await?;
        Ok(())
    }

    async fn create_stream(
        &self,
        id: &SessionId,
        message: ClientJsonRpcMessage,
    ) -> Result<impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static, Self::Error>
    {
        Ok(self.inner.create_stream(id, message).await?)
    }

    async fn accept_message(
        &self,
        id: &SessionId,
        message: ClientJsonRpcMessage,
    ) -> Result<(), Self::Error> {
        Ok(self.inner.accept_message(id, message).await?)
    }

    async fn create_standalone_stream(
        &self,
        id: &SessionId,
    ) -> Result<impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static, Self::Error>
    {
        Ok(self.inner.create_standalone_stream(id).await?)
    }

    async fn resume(
        &self,
        id: &SessionId,
        last_event_id: String,
    ) -> Result<impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static, Self::Error>
    {
        Ok(self.inner.resume(id, last_event_id).await?)
    }
}

/// Axum middleware that rewrites `rmcp`'s generic `500 Internal Server Error`
/// into `429 Too Many Requests` when the failure was
/// [`CappedSessionManagerError::CapReached`] (detected via
/// [`SESSION_CAP_MARKER`] in the response body), adding a `Retry-After`
/// header.
///
/// This runs as response post-processing rather than a request pre-check
/// because only the real [`SessionManager::create_session`] call — deep
/// inside `rmcp` — knows whether a given request actually attempts to create
/// a session; see [`CappedSessionManager`]'s docs for why that can't be
/// determined from the request alone.
#[cfg(feature = "transport-http")]
async fn enforce_session_cap(
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    let response = next.run(request).await;
    if response.status() != axum::http::StatusCode::INTERNAL_SERVER_ERROR {
        return response;
    }

    let (mut parts, body) = response.into_parts();
    // `create_session` failures always render as a small `Full<Bytes>` body
    // (`internal_error_response` in rmcp's `server_side_http.rs`); the large
    // streaming SSE/JSON success bodies never carry a 500 status, so this
    // never touches them. 64 KiB is far beyond any realistic error message.
    let Ok(bytes) = axum::body::to_bytes(body, 64 * 1024).await else {
        // Buffering the original error body failed (e.g. it exceeded the 64
        // KiB cap, which should never happen per the comment above, or the
        // body stream errored). Preserve the 500 status but substitute a
        // minimal fallback body rather than dropping the error entirely.
        return axum::response::Response::from_parts(
            parts,
            axum::body::Body::from("Internal Server Error"),
        );
    };

    if bytes
        .windows(SESSION_CAP_MARKER.len())
        .any(|window| window == SESSION_CAP_MARKER.as_bytes())
    {
        parts.status = axum::http::StatusCode::TOO_MANY_REQUESTS;
        parts.headers.insert(
            axum::http::header::RETRY_AFTER,
            axum::http::HeaderValue::from_static("1"),
        );
        return axum::response::Response::from_parts(
            parts,
            axum::body::Body::from("Too Many Requests: maximum concurrent HTTP sessions reached"),
        );
    }

    axum::response::Response::from_parts(parts, axum::body::Body::from(bytes))
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    /// `Transport::Stdio` is always constructible regardless of feature flags.
    #[test]
    fn test_transport_stdio_variant() {
        let t = super::Transport::Stdio;
        assert!(matches!(t, super::Transport::Stdio));
    }

    /// #329 regression: `crate::shutdown`'s cleanup-window fix hinges on a
    /// freshly constructed `ShutdownSignal` still receiving real OS signals
    /// after an *earlier* `ShutdownSignal` (e.g. the one `run_stdio`/
    /// `run_http` held) has already been dropped — proving there is no
    /// window in which the OS handler itself gets deregistered (per
    /// `ShutdownSignal`'s corrected struct doc: `tokio::signal` never
    /// uninstalls it, regardless of how many instances are constructed or
    /// dropped). Exercises this with a real self-sent `SIGTERM` via the
    /// external `kill` binary rather than mocking `ShutdownSignal`, since
    /// that's the exact mechanism `crate::shutdown`'s force-exit task relies
    /// on. Safe under `cargo nextest`'s one-process-per-test model, so no
    /// other test's signal disposition is affected.
    ///
    /// Deliberately does not go through `crate::shutdown` itself: a signal
    /// caught there unconditionally calls `std::process::exit(1)`, which
    /// would kill this test's own process for real rather than fail an
    /// assertion — see the #329 regression-test handoff for why a test
    /// triggering `std::process::exit(1)` isn't attempted here.
    ///
    /// Unix-only: `SIGTERM` and the external `kill` binary this test relies
    /// on don't exist on Windows, where `ShutdownSignal` listens for
    /// Ctrl-C instead (see the struct's `#[cfg(windows)]` arm above).
    #[cfg(unix)]
    #[tokio::test]
    async fn test_fresh_shutdown_signal_still_receives_sigterm_after_prior_instance_dropped() {
        let earlier = super::ShutdownSignal::new();
        drop(earlier);

        let mut cleanup_signal = super::ShutdownSignal::new();

        let pid = std::process::id();
        let signal_sender = tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            let status = std::process::Command::new("kill")
                .arg("-TERM")
                .arg(pid.to_string())
                .status()
                .unwrap();
            assert!(status.success(), "`kill -TERM {pid}` must succeed");
        });

        let result =
            tokio::time::timeout(std::time::Duration::from_secs(5), cleanup_signal.recv()).await;
        signal_sender.await.unwrap();

        assert!(
            result.is_ok(),
            "a freshly constructed ShutdownSignal must still receive a real SIGTERM sent after \
             an earlier instance was dropped — this is the exact mechanism crate::shutdown's \
             cleanup-window force-exit task depends on"
        );
    }

    /// #241: `run_stdio` must not hang when the transport never even
    /// establishes — it must surface the failure promptly.
    ///
    /// This is the closest portable coverage of `run_stdio`'s non-signal
    /// path achievable here: `run_stdio` is hardcoded to the process's real
    /// stdin/stdout (no injectable transport), and this crate is
    /// `deny(unsafe_code)`, so a test can't redirect the fd to simulate "the
    /// MCP handshake completes, *then* stdin closes" — the specific
    /// scenario that would drive `service.waiting()` to resolve inside the
    /// `tokio::select!` and hit its `Ok(())` arm. What a test *can* rely on:
    /// under `cargo nextest`, each test's stdin is already closed before the
    /// test body runs, so `mcp_server.serve(...)` fails during the initial
    /// `initialize` handshake — before `run_stdio` ever reaches the
    /// `select!`. That still exercises real production code (the `.serve()`
    /// call and its error mapping) and proves `run_stdio` returns promptly
    /// rather than hanging, which is what a broken `select!` (e.g. one
    /// missing a branch, or awaiting the wrong future) would look like.
    #[tokio::test]
    async fn test_run_stdio_returns_promptly_when_stdin_is_already_closed() {
        use std::path::PathBuf;
        use std::sync::Arc;

        use tokio::sync::Mutex;

        use crate::bridge::{NotificationCache, SubscriptionRegistry, Translator};
        use crate::config::McpConfig;
        use crate::mcp::McplsServer;

        let translator = Arc::new(Translator::new());
        let notification_cache = Arc::new(Mutex::new(NotificationCache::new()));
        let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new());
        let subs = SubscriptionRegistry::new();
        let server = McplsServer::new(
            translator,
            notification_cache,
            workspace_roots,
            subs,
            false,
            McpConfig::default(),
        );
        let peer_cell = tokio::sync::OnceCell::new();

        let outcome = tokio::time::timeout(
            std::time::Duration::from_secs(2),
            super::run_stdio(server, &peer_cell, super::ShutdownSignal::new()),
        )
        .await;

        assert!(
            outcome.is_ok(),
            "run_stdio must not hang when stdin is already closed"
        );
        let result = outcome.unwrap();
        assert!(
            matches!(result, Err(crate::Error::McpServer(_))),
            "expected a McpServer error from the failed handshake, got: {result:?}"
        );
    }

    #[cfg(feature = "transport-http")]
    mod http_tests {
        use std::net::SocketAddr;

        use super::super::{HttpConfig, Transport};
        use crate::test_lsp::CapturedLogs;

        #[test]
        fn test_http_config_fields() {
            let addr: SocketAddr = "127.0.0.1:3000".parse().unwrap();
            let cfg = HttpConfig::new(addr, "/mcp");
            assert_eq!(cfg.bind, addr);
            assert_eq!(cfg.path, "/mcp");
        }

        #[test]
        fn test_http_config_clone() {
            let cfg = HttpConfig::new("127.0.0.1:3001".parse().unwrap(), "/test");
            let cloned = cfg.clone();
            assert_eq!(cloned.bind, cfg.bind);
            assert_eq!(cloned.path, cfg.path);
        }

        #[test]
        fn test_transport_http_variant() {
            let cfg = HttpConfig::new("127.0.0.1:3002".parse().unwrap(), "/mcp");
            let t = Transport::Http(cfg);
            assert!(matches!(t, Transport::Http(_)));
        }

        #[test]
        fn test_http_config_new_uses_default_limits() {
            let cfg = HttpConfig::new("127.0.0.1:3003".parse().unwrap(), "/mcp");
            assert_eq!(
                cfg.max_request_body_bytes,
                HttpConfig::DEFAULT_MAX_REQUEST_BODY_BYTES
            );
            assert_eq!(
                cfg.max_concurrent_sessions,
                HttpConfig::DEFAULT_MAX_CONCURRENT_SESSIONS
            );
        }

        #[test]
        fn test_http_config_with_max_request_body_bytes_overrides_default() {
            let cfg = HttpConfig::new("127.0.0.1:3004".parse().unwrap(), "/mcp")
                .with_max_request_body_bytes(1024);
            assert_eq!(cfg.max_request_body_bytes, 1024);
            assert_eq!(
                cfg.max_concurrent_sessions,
                HttpConfig::DEFAULT_MAX_CONCURRENT_SESSIONS
            );
        }

        #[test]
        fn test_http_config_with_max_concurrent_sessions_overrides_default() {
            let cfg = HttpConfig::new("127.0.0.1:3005".parse().unwrap(), "/mcp")
                .with_max_concurrent_sessions(5);
            assert_eq!(cfg.max_concurrent_sessions, 5);
            assert_eq!(
                cfg.max_request_body_bytes,
                HttpConfig::DEFAULT_MAX_REQUEST_BODY_BYTES
            );
        }

        /// Verifies `run_http` binds successfully and accepts TCP connections.
        #[tokio::test]
        async fn test_run_http_binds() {
            use std::path::PathBuf;
            use std::sync::Arc;

            use tokio::sync::Mutex;

            use crate::bridge::{NotificationCache, SubscriptionRegistry, Translator};
            use crate::config::McpConfig;
            use crate::mcp::McplsServer;

            let translator = Arc::new(Translator::new());
            let notification_cache = Arc::new(Mutex::new(NotificationCache::new()));
            let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new());
            let subs = SubscriptionRegistry::new();
            let server = McplsServer::new(
                translator,
                notification_cache,
                workspace_roots,
                subs,
                false,
                McpConfig::default(),
            );

            // Bind port 0 so the OS assigns a free port.
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp");

            let server_task = tokio::spawn(super::super::run_http(
                server,
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            // A successful TCP connect proves the listener is up.
            let connected = tokio::net::TcpStream::connect(addr).await;
            assert!(
                connected.is_ok(),
                "HTTP listener should accept TCP connections"
            );

            server_task.abort();
        }

        /// #241 C1 regression: `run_http` must not self-terminate after
        /// `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` of ordinary uptime when no
        /// shutdown signal has been sent — the graceful-shutdown timeout
        /// must only start counting once a signal actually arrives, not
        /// from server startup.
        ///
        /// Uses `#[tokio::test(start_paused = true)]` plus
        /// `tokio::time::advance` to fast-forward virtual time past the
        /// timeout instead of sleeping the real 30s. Under the bug this
        /// regresses against — `tokio::time::timeout(HTTP_GRACEFUL_SHUTDOWN_TIMEOUT,
        /// serve)` wrapping the whole `serve` future from construction —
        /// advancing virtual time past the timeout resolves that timer and
        /// finishes the task immediately, even with no signal sent. Under
        /// the fix, nothing inside `run_http` starts a timer until `cancel`
        /// is cancelled, so this advance must have no effect and the task
        /// must still be running.
        #[tokio::test(start_paused = true)]
        async fn test_run_http_does_not_self_terminate_without_signal() {
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp");
            let server_task = tokio::spawn(super::super::run_http(
                test_server(),
                cfg,
                super::super::ShutdownSignal::new(),
            ));

            // Let the spawned task make initial progress (bind the
            // listener, enter its `select!`) without depending on any real
            // or virtual delay.
            for _ in 0..10 {
                tokio::task::yield_now().await;
            }

            // Fast-forward well past `HTTP_GRACEFUL_SHUTDOWN_TIMEOUT` with
            // no shutdown signal ever sent.
            tokio::time::advance(
                super::super::HTTP_GRACEFUL_SHUTDOWN_TIMEOUT + std::time::Duration::from_secs(5),
            )
            .await;
            for _ in 0..10 {
                tokio::task::yield_now().await;
            }

            assert!(
                !server_task.is_finished(),
                "run_http must still be serving after HTTP_GRACEFUL_SHUTDOWN_TIMEOUT of uptime \
                 with no shutdown signal sent"
            );

            server_task.abort();
        }

        /// Verifies `run_http` returns an error when the bind address is already in use.
        #[tokio::test]
        async fn test_run_http_bind_error() {
            use std::path::PathBuf;
            use std::sync::Arc;

            use tokio::sync::Mutex;

            use crate::bridge::{NotificationCache, SubscriptionRegistry, Translator};
            use crate::config::McpConfig;
            use crate::mcp::McplsServer;

            // Hold a listener to make the port unavailable.
            let occupied = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = occupied.local_addr().unwrap();

            let translator = Arc::new(Translator::new());
            let notification_cache = Arc::new(Mutex::new(NotificationCache::new()));
            let workspace_roots: Arc<[PathBuf]> = Arc::from(Vec::new());
            let subs = SubscriptionRegistry::new();
            let server = McplsServer::new(
                translator,
                notification_cache,
                workspace_roots,
                subs,
                false,
                McpConfig::default(),
            );

            let cfg = HttpConfig::new(addr, "/mcp");

            let result =
                super::super::run_http(server, cfg, super::super::ShutdownSignal::new()).await;
            assert!(
                result.is_err(),
                "run_http should fail when port is occupied"
            );

            drop(occupied);
        }

        /// Builds a `McplsServer` with default collaborators and the given
        /// workspace roots, matching the setup shared by every
        /// `run_http`-driving test in this module.
        fn test_server_with_roots(
            workspace_roots: std::sync::Arc<[std::path::PathBuf]>,
        ) -> crate::mcp::McplsServer {
            use std::sync::Arc;

            use tokio::sync::Mutex;

            use crate::bridge::{NotificationCache, SubscriptionRegistry, Translator};
            use crate::config::McpConfig;
            use crate::mcp::McplsServer;

            let translator = Arc::new(Translator::new());
            let notification_cache = Arc::new(Mutex::new(NotificationCache::new()));
            let subs = SubscriptionRegistry::new();
            McplsServer::new(
                translator,
                notification_cache,
                workspace_roots,
                subs,
                false,
                McpConfig::default(),
            )
        }

        /// [`test_server_with_roots`] with no workspace roots configured.
        fn test_server() -> crate::mcp::McplsServer {
            test_server_with_roots(std::sync::Arc::from(Vec::new()))
        }

        /// Sends a raw HTTP/1.1 POST request over TCP and returns the raw response
        /// text (status line, headers, and body). Used because neither `reqwest`
        /// nor `tower`/`http-body-util` are available as dev-dependencies here.
        async fn raw_http_post(
            addr: SocketAddr,
            path: &str,
            extra_headers: &str,
            body: &[u8],
        ) -> String {
            use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

            let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
            let request = format!(
                "POST {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n{extra_headers}Content-Length: {}\r\n\r\n",
                body.len()
            );
            stream.write_all(request.as_bytes()).await.unwrap();
            stream.write_all(body).await.unwrap();

            let mut response = Vec::new();
            let mut buf = [0u8; 8192];
            loop {
                match tokio::time::timeout(std::time::Duration::from_secs(2), stream.read(&mut buf))
                    .await
                {
                    Ok(Ok(0)) | Err(_) => break,
                    Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
                    Ok(Err(e)) => panic!("read error: {e}"),
                }
            }
            String::from_utf8_lossy(&response).into_owned()
        }

        /// A POST body exceeding `cfg.max_request_body_bytes` must be rejected
        /// with `413 Payload Too Large`, proving the config value reaches
        /// `StreamableHttpServerConfig::max_request_body_bytes`.
        #[tokio::test]
        async fn test_run_http_rejects_oversized_body_with_413() {
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp").with_max_request_body_bytes(64);
            let server_task = tokio::spawn(super::super::run_http(
                test_server(),
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let oversized_body = vec![b'a'; 65];
            let response = raw_http_post(
                addr,
                "/mcp",
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\n",
                &oversized_body,
            )
            .await;

            assert!(
                response.starts_with("HTTP/1.1 413"),
                "expected 413 Payload Too Large, got: {response}"
            );

            server_task.abort();
        }

        /// A POST body within `cfg.max_request_body_bytes` must not be rejected
        /// for size — it reaches JSON deserialization instead (the body here is
        /// intentionally not valid JSON-RPC, so a non-413 error distinguishes
        /// "passed the size check" from "was a valid request").
        #[tokio::test]
        async fn test_run_http_accepts_body_within_limit() {
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp").with_max_request_body_bytes(64);
            let server_task = tokio::spawn(super::super::run_http(
                test_server(),
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let small_body = vec![b'a'; 32];
            let response = raw_http_post(
                addr,
                "/mcp",
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\n",
                &small_body,
            )
            .await;

            assert!(
                !response.starts_with("HTTP/1.1 413"),
                "body within limit must not be rejected as too large, got: {response}"
            );

            server_task.abort();
        }

        /// `CappedSessionManager::create_session` must enforce a hard bound:
        /// once `max_sessions` sessions exist, the next `create_session` call
        /// fails with the capacity marker, and closing a session frees the
        /// slot back up for a subsequent `create_session` to succeed.
        // `manager` is used until the end of the test — clippy's drop-tightening
        // heuristic misreads that as an early-droppable temporary because
        // `CappedSessionManager` embeds a `tokio::sync::Mutex`.
        #[allow(clippy::significant_drop_tightening)]
        #[tokio::test]
        async fn test_capped_session_manager_enforces_hard_bound() {
            use rmcp::transport::streamable_http_server::session::SessionManager as _;

            let manager = super::super::CappedSessionManager::new(1);

            let (first_id, _transport) = manager.create_session().await.unwrap();

            let second_err = manager.create_session().await.map(|_| ()).unwrap_err();
            assert!(
                matches!(
                    second_err,
                    super::super::CappedSessionManagerError::CapReached
                ),
                "expected CapReached once at capacity, got: {second_err:?}"
            );

            manager.close_session(&first_id).await.unwrap();

            let (third_id, _transport) = manager.create_session().await.unwrap();
            assert_ne!(first_id, third_id);
        }

        /// Regression guard for S2: concurrent `create_session` calls must not
        /// overshoot `max_sessions`. Unlike the sequential test above (which
        /// would pass even against a racy check-then-create implementation),
        /// this spawns `N > max_sessions` calls at once and asserts exactly
        /// `max_sessions` succeed — the one test shape that actually
        /// distinguishes the atomic-semaphore design from a TOCTOU race.
        // `manager` is used until the end of the test — see the identical
        // drop-tightening note on `test_capped_session_manager_enforces_hard_bound`.
        #[allow(clippy::significant_drop_tightening)]
        #[tokio::test]
        async fn test_capped_session_manager_bounds_concurrent_create_session() {
            use rmcp::transport::streamable_http_server::session::SessionManager as _;

            const MAX_SESSIONS: usize = 5;
            const CONCURRENT_ATTEMPTS: usize = 25;

            let manager =
                std::sync::Arc::new(super::super::CappedSessionManager::new(MAX_SESSIONS));

            let mut tasks = tokio::task::JoinSet::new();
            for _ in 0..CONCURRENT_ATTEMPTS {
                let manager = manager.clone();
                tasks.spawn(async move { manager.create_session().await.is_ok() });
            }

            let mut succeeded = 0usize;
            while let Some(result) = tasks.join_next().await {
                if result.unwrap() {
                    succeeded += 1;
                }
            }

            assert_eq!(
                succeeded, MAX_SESSIONS,
                "exactly max_sessions concurrent create_session calls must succeed"
            );
        }

        /// Narrower unit test of the `enforce_session_cap` middleware itself
        /// (rather than the full `run_http` wiring): a `500` response whose
        /// body carries `SESSION_CAP_MARKER` must be rewritten to `429` with
        /// a `Retry-After` header.
        #[tokio::test]
        async fn test_enforce_session_cap_rewrites_capacity_marker_to_429() {
            let app = axum::Router::new()
                .route(
                    "/",
                    axum::routing::post(|| async {
                        (
                            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
                            format!(
                                "Encounter an error when create session: {}: maximum concurrent \
                                 HTTP sessions already active",
                                super::super::SESSION_CAP_MARKER
                            ),
                        )
                    }),
                )
                .layer(axum::middleware::from_fn(super::super::enforce_session_cap));

            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let server_task = tokio::spawn(async move {
                axum::serve(listener, app).await.unwrap();
            });
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let response = raw_http_post(addr, "/", "", b"{}").await;
            assert!(
                response.starts_with("HTTP/1.1 429"),
                "expected 429 for a marker-carrying 500, got: {response}"
            );
            assert!(
                response.to_lowercase().contains("retry-after"),
                "expected a Retry-After header, got: {response}"
            );

            server_task.abort();
        }

        /// A `500` response whose body does *not* carry `SESSION_CAP_MARKER`
        /// (an unrelated internal error) must pass through unchanged, proving
        /// the middleware doesn't misclassify every `500` as a capacity
        /// rejection.
        #[tokio::test]
        async fn test_enforce_session_cap_leaves_unrelated_500_untouched() {
            let app = axum::Router::new()
                .route(
                    "/",
                    axum::routing::post(|| async {
                        (
                            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
                            "Encounter an error when create session: some unrelated failure",
                        )
                    }),
                )
                .layer(axum::middleware::from_fn(super::super::enforce_session_cap));

            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let server_task = tokio::spawn(async move {
                axum::serve(listener, app).await.unwrap();
            });
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let response = raw_http_post(addr, "/", "", b"{}").await;
            assert!(
                response.starts_with("HTTP/1.1 500"),
                "unrelated 500s must not be rewritten to 429, got: {response}"
            );

            server_task.abort();
        }

        /// End-to-end: with `max_concurrent_sessions(1)`, a second concurrent
        /// `initialize` handshake over `run_http` must be rejected with `429`
        /// once the first session is established.
        #[tokio::test]
        async fn test_run_http_rejects_new_session_at_capacity_with_429() {
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp").with_max_concurrent_sessions(1);
            let server_task = tokio::spawn(super::super::run_http(
                test_server(),
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let initialize_body = br#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}"#;
            let accept_headers =
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\n";

            // First handshake must succeed and establish a session.
            let first = raw_http_post(addr, "/mcp", accept_headers, initialize_body).await;
            assert!(
                first.starts_with("HTTP/1.1 200"),
                "first initialize handshake should succeed, got: {first}"
            );

            // Second handshake, with the sole slot still held, must be capped.
            let second = raw_http_post(addr, "/mcp", accept_headers, initialize_body).await;
            assert!(
                second.starts_with("HTTP/1.1 429"),
                "second initialize handshake should be rejected once at capacity, got: {second}"
            );

            server_task.abort();
        }

        /// S1 non-regression: a non-`initialize` request carrying SEP-2575
        /// per-request `_meta` protocol-version metadata
        /// (`io.modelcontextprotocol/protocolVersion` = `2026-07-28` plus the
        /// required `clientCapabilities` key) takes rmcp 3.2.0's stateless
        /// discover-lifecycle path and never calls
        /// `SessionManager::create_session` — `rmcp` serves it directly
        /// without touching the session table — so it must not be rejected
        /// by the cap even while `max_concurrent_sessions` legacy sessions
        /// are already active. (An `initialize` request is always
        /// classified legacy in 3.2.0 regardless of the protocol version it
        /// names, so it cannot be used to probe the stateless path.) This
        /// guards against a future refactor reintroducing request-header
        /// sniffing for the cap decision (the bug this design replaced).
        #[tokio::test]
        async fn test_run_http_stateless_request_bypasses_session_cap() {
            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let cfg = HttpConfig::new(addr, "/mcp").with_max_concurrent_sessions(1);
            let server_task = tokio::spawn(super::super::run_http(
                test_server(),
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let accept_headers =
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\n";

            // Fill the sole legacy-session slot.
            let legacy_initialize = br#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}"#;
            let legacy = raw_http_post(addr, "/mcp", accept_headers, legacy_initialize).await;
            assert!(
                legacy.starts_with("HTTP/1.1 200"),
                "legacy initialize should succeed and consume the sole session slot, got: {legacy}"
            );

            // A non-`initialize` request carrying per-request `_meta`
            // protocol-version metadata takes rmcp 3.2.0's stateless
            // discover-lifecycle path and never creates a session, so it
            // must bypass the cap entirely even though the slot above is
            // still held. The `MCP-Protocol-Version` header must match the
            // `_meta` value once the latter is present, and declaring
            // `2026-07-28` also brings in SEP-2243's `Mcp-Method` header
            // requirement.
            let stateless_headers = "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMCP-Protocol-Version: 2026-07-28\r\nMcp-Method: resources/list\r\n";
            let stateless_request = br#"{"jsonrpc":"2.0","id":2,"method":"resources/list","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}"#;
            let stateless = raw_http_post(addr, "/mcp", stateless_headers, stateless_request).await;
            assert!(
                stateless.starts_with("HTTP/1.1 200"),
                "stateless requests must bypass the session cap entirely, got: {stateless}"
            );

            server_task.abort();
        }

        /// #478/#482 regression pin: on rmcp's stateless path (see
        /// `test_run_http_stateless_request_bypasses_session_cap` above), the
        /// service factory -- and therefore `McplsServer::for_new_session` --
        /// runs once per *request*, not once per session, registering a new
        /// `ResourceSubscriptions` set each time. Firing many stateless
        /// requests must not grow `SubscriptionRegistry` without bound (S2's
        /// prune-before-push in `register`); it must stay pinned near the
        /// handful of instances actually alive (the long-lived template
        /// `McplsServer` this test built, plus at most one not-yet-pruned
        /// stateless-request entry).
        #[tokio::test]
        async fn test_stateless_requests_do_not_grow_subscription_registry_unboundedly() {
            const REQUEST_COUNT: u32 = 20;

            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let server = test_server();
            let registry = server.subscription_registry();

            let cfg = HttpConfig::new(addr, "/mcp");
            let server_task = tokio::spawn(super::super::run_http(
                server,
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let stateless_headers = "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMCP-Protocol-Version: 2026-07-28\r\nMcp-Method: resources/list\r\n";

            for id in 0..REQUEST_COUNT {
                let body = format!(
                    r#"{{"jsonrpc":"2.0","id":{id},"method":"resources/list","params":{{"_meta":{{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}"#
                );
                let response =
                    raw_http_post(addr, "/mcp", stateless_headers, body.as_bytes()).await;
                assert!(
                    response.starts_with("HTTP/1.1 200"),
                    "stateless request {id} should succeed, got: {response}"
                );
            }

            // Let the background task each stateless request's handler is
            // spawned on (`serve_directly_with_ct`'s `waiting()` task in rmcp)
            // finish dropping its per-request `McplsServer`.
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            // Raw (unpruned) length: nothing in this test ever calls
            // `any_contains`/`is_all_empty` (no LSP server is running to
            // publish diagnostics), so the only thing that can keep this
            // bounded is `register` pruning dead entries on the way in.
            let raw = registry.raw_len();
            assert!(
                raw <= 2,
                "registry should stay bounded across {REQUEST_COUNT} stateless requests, got {raw} raw entries"
            );

            server_task.abort();
        }

        /// Shared setup for the `#482` regression tests below: a real
        /// `run_http` server with one file inside its sole workspace root, so
        /// `resources/subscribe` requests validate and reach the handler.
        struct SubscribeTestServer {
            addr: SocketAddr,
            uri: String,
            server_task: tokio::task::JoinHandle<Result<(), crate::Error>>,
            // Held so the file `subscribe` canonicalizes stays on disk.
            _workspace: tempfile::TempDir,
        }

        async fn spawn_subscribe_test_server() -> SubscribeTestServer {
            let workspace = tempfile::TempDir::new().unwrap();
            let file_path = workspace.path().join("main.rs");
            std::fs::write(&file_path, "fn main() {}").unwrap();
            let uri = crate::bridge::resources::make_uri(&file_path).unwrap();

            let probe = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = probe.local_addr().unwrap();
            drop(probe);

            let server =
                test_server_with_roots(std::sync::Arc::from(vec![workspace.path().to_path_buf()]));

            let cfg = HttpConfig::new(addr, "/mcp");
            let server_task = tokio::spawn(super::super::run_http(
                server,
                cfg,
                super::super::ShutdownSignal::new(),
            ));
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            SubscribeTestServer {
                addr,
                uri,
                server_task,
                _workspace: workspace,
            }
        }

        /// #482: `_meta` negotiating `2026-07-28` per request is stateless by
        /// both rmcp and mcpls' reckoning, so rmcp itself answers
        /// `-32601 method not found` before dispatch.
        #[tokio::test]
        async fn test_stateless_subscribe_negotiated_per_request_is_rejected_by_rmcp() {
            let srv = spawn_subscribe_test_server().await;
            let uri = &srv.uri;

            let headers = format!(
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMCP-Protocol-Version: 2026-07-28\r\nMcp-Method: resources/subscribe\r\nMcp-Name: {uri}\r\n"
            );
            let body = format!(
                r#"{{"jsonrpc":"2.0","id":1,"method":"resources/subscribe","params":{{"uri":"{uri}","_meta":{{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}"#
            );
            let response = raw_http_post(srv.addr, "/mcp", &headers, body.as_bytes()).await;
            assert!(
                response.contains("-32601"),
                "expected rmcp to refuse dispatching a 2026-07-28-negotiated resources/subscribe \
                 (method not found), got: {response}"
            );

            srv.server_task.abort();
        }

        /// #482 regression matrix: `_meta` naming a pre-`2026-07-28` version
        /// (see [`super::super::request_uses_discover_lifecycle_meta`]'s docs
        /// for why rmcp still serves this statelessly) must be rejected by
        /// mcpls' own guard for both `subscribe` and `unsubscribe`, and a
        /// fabricated `Mcp-Session-Id` must not bypass it.
        #[tokio::test]
        async fn test_stateless_lifecycle_mismatch_is_rejected_by_mcpls() {
            let srv = spawn_subscribe_test_server().await;
            let uri = &srv.uri;

            for (method, extra_header) in [
                ("resources/subscribe", ""),
                (
                    "resources/subscribe",
                    "Mcp-Session-Id: not-a-real-session\r\n",
                ),
                ("resources/unsubscribe", ""),
            ] {
                let headers = format!(
                    "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMCP-Protocol-Version: 2025-06-18\r\n{extra_header}"
                );
                let body = format!(
                    r#"{{"jsonrpc":"2.0","id":1,"method":"{method}","params":{{"uri":"{uri}","_meta":{{"io.modelcontextprotocol/protocolVersion":"2025-06-18","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}"#
                );
                let response = raw_http_post(srv.addr, "/mcp", &headers, body.as_bytes()).await;
                assert!(
                    response.contains("-32052") && !response.contains(r#""result":{}"#),
                    "{method} (extra header: {extra_header:?}) must be rejected by mcpls' \
                     stateless-subscription guard, not silently succeed, got: {response}"
                );
            }

            srv.server_task.abort();
        }

        /// #482 non-regression: a legacy session's `resources/subscribe`,
        /// sent with the session's assigned `Mcp-Session-Id` echoed back and
        /// no per-request `_meta`, must not be rejected by either guard above.
        #[tokio::test]
        async fn test_legacy_session_subscribe_is_not_rejected_as_stateless() {
            let srv = spawn_subscribe_test_server().await;
            let uri = &srv.uri;
            let session_id = initialize_legacy_session(srv.addr).await;

            let session_headers = format!(
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMcp-Session-Id: {session_id}\r\n"
            );
            let subscribe_body = format!(
                r#"{{"jsonrpc":"2.0","id":2,"method":"resources/subscribe","params":{{"uri":"{uri}"}}}}"#
            );
            let session_response = raw_http_post(
                srv.addr,
                "/mcp",
                &session_headers,
                subscribe_body.as_bytes(),
            )
            .await;
            assert!(
                !session_response.contains("-32601") && !session_response.contains("-32052"),
                "a legacy session's subscribe must not be rejected as stateless, got: \
                 {session_response}"
            );

            srv.server_task.abort();
        }

        /// #482: a live session's `resources/subscribe` that also carries
        /// per-request discover-lifecycle `_meta` is rejected too -- rmcp
        /// serves it statelessly regardless of the session id, so this is
        /// intentional, not a regression.
        #[tokio::test]
        async fn test_legacy_session_subscribe_with_discover_meta_is_rejected_as_stateless() {
            let srv = spawn_subscribe_test_server().await;
            let uri = &srv.uri;
            let session_id = initialize_legacy_session(srv.addr).await;

            let headers = format!(
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\nMcp-Session-Id: {session_id}\r\nMCP-Protocol-Version: 2025-06-18\r\n"
            );
            let body = format!(
                r#"{{"jsonrpc":"2.0","id":2,"method":"resources/subscribe","params":{{"uri":"{uri}","_meta":{{"io.modelcontextprotocol/protocolVersion":"2025-06-18","io.modelcontextprotocol/clientCapabilities":{{}}}}}}}}"#
            );
            let response = raw_http_post(srv.addr, "/mcp", &headers, body.as_bytes()).await;
            assert!(
                response.contains("-32052"),
                "a live session's subscribe with per-request discover _meta must still be \
                 rejected -- rmcp serves it statelessly regardless of the session id, got: \
                 {response}"
            );

            srv.server_task.abort();
        }

        /// Performs the `initialize` handshake for a legacy HTTP session and
        /// returns its assigned `Mcp-Session-Id`.
        async fn initialize_legacy_session(addr: SocketAddr) -> String {
            let accept_headers =
                "Accept: application/json, text/event-stream\r\nContent-Type: application/json\r\n";
            let initialize_body = br#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}"#;
            let init_response = raw_http_post(addr, "/mcp", accept_headers, initialize_body).await;
            assert!(
                init_response.starts_with("HTTP/1.1 200"),
                "legacy initialize should succeed, got: {init_response}"
            );
            init_response
                .lines()
                .find_map(|line| {
                    let (name, value) = line.split_once(':')?;
                    name.eq_ignore_ascii_case("mcp-session-id")
                        .then(|| value.trim().to_string())
                })
                .unwrap()
        }

        /// #233: binding to a non-loopback address must log a warning that
        /// tells operators to put the endpoint behind a reverse proxy that
        /// *enforces* authentication (not the inverted "ensure no
        /// authentication is required" wording it replaced).
        #[tokio::test]
        async fn test_run_http_non_loopback_bind_warns_to_use_reverse_proxy() {
            use tracing_subscriber::layer::SubscriberExt as _;

            let addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
            let cfg = HttpConfig::new(addr, "/mcp");

            let captured = CapturedLogs::default();
            let subscriber = tracing_subscriber::registry().with(captured.clone());
            let guard = tracing::subscriber::set_default(subscriber);

            // The warning fires synchronously right after bind, before
            // `axum::serve` starts running indefinitely, so a short timeout
            // is enough to observe it.
            let _ = tokio::time::timeout(
                std::time::Duration::from_millis(200),
                super::super::run_http(test_server(), cfg, super::super::ShutdownSignal::new()),
            )
            .await;

            drop(guard);

            let messages = captured.messages();
            assert!(
                messages.iter().any(|m| m.contains(
                    "place this endpoint behind a reverse proxy that enforces authentication"
                )),
                "expected reverse-proxy warning in captured tracing events, got: {messages:?}"
            );
        }
    }
}