//! Server connector.
//!
//! Bridges a typed [`ServerSpec`] from the parsed suite to a live
//! [`Client`]. The
//! transport layer already ships per-protocol constructors; the
//! connector picks the right one, spins it up, runs the MCP
//! `initialize` handshake, and hands back a connected client plus a
//! handle the runner uses for shutdown.
//!
//! The connector wires stdio, streamable-http, legacy-sse, and cassette
//! replay. For URL servers it injects a cached OAuth access token from
//! `mcptest login` when the spec did not set an explicit Authorization header.
//! The other transport-level auth schemes (SigV4, mTLS, Web Bot Auth) remain
//! follow-up tickets; the seam exists so they land additively.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use tokio::sync::Mutex;

use crate::auth::{Refresher, ReqwestRefresher, ReqwestRefresherConfig, Token, TokenCache};
use crate::protocol::{Client, ClientCapabilities, ProtocolError, ProtocolVersion};
use crate::suite::{OAuthClientCredentials, ServerSpec};
use crate::transport::{
    LegacySse, LegacySseConfig, LegacySseError, StdioConfig, StdioError, StdioTransport,
    StreamableHttp, StreamableHttpConfig, StreamableHttpError,
};

/// Default per-request timeout the connector applies when the caller
/// does not override it. Lines up with the transport-level default so
/// the operator sees consistent timeout behavior whether the failure
/// is on the wire or the server.
pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);

/// Window for diagnosing a failed stdio handshake as a child exit
/// (WOR-1326). A child that died before answering `initialize` is
/// usually reapable within milliseconds of the handshake error; one
/// second covers slow CI boxes without delaying genuinely live-but-
/// broken servers by more than that.
const STARTUP_EXIT_GRACE: Duration = Duration::from_secs(1);

/// Live MCP connection: the [`Client`] is what the executor calls;
/// the [`Shutdown`] is what the runner drops to tear the connection
/// down deterministically at end-of-run.
pub struct LiveServer {
    /// Connected client, post-`initialize`. Ready to receive
    /// `tools/call`, `resources/read`, `prompts/get`, etc.
    pub client: Client,
    /// Transport-specific shutdown handle. Drop or call
    /// [`Shutdown::shutdown`] to close the wire.
    pub shutdown: Shutdown,
    /// Per-request response header capture (WOR-1287). `Some` only for
    /// URL transports (streamable HTTP, legacy SSE); stdio and cassette
    /// targets have no HTTP responses, so the executor's header gate can
    /// tell "no headers on this transport" from "headers not captured".
    pub header_capture: Option<crate::transport::HeaderCapture>,
}

/// Transport-specific shutdown handle. Stored as an enum so a single
/// `HashMap<String, LiveServer>` can hold mixed-transport entries.
///
/// The variants wrap transport handles of differing size (the stdio child
/// handle is larger on Windows, which trips `large_enum_variant` only there).
/// There is exactly one `Shutdown` per connected server, so the size of the
/// largest variant is not a concern worth boxing every transport for; allow
/// the lint rather than complicate the shutdown call sites.
#[allow(clippy::large_enum_variant)]
pub enum Shutdown {
    /// Stdio subprocess: dropping the handle kills the child via
    /// `kill_on_drop`; calling `shutdown` is the graceful path.
    Stdio(Mutex<Option<StdioTransport>>),
    /// Streamable HTTP: aborts background reader tasks.
    StreamableHttp(Mutex<Option<StreamableHttp>>),
    /// Legacy SSE: aborts background reader tasks.
    LegacySse(Mutex<Option<LegacySse>>),
    /// Cassette replay: aborts the background replay task. Holds no OS
    /// resources, so this is purely for symmetry with the live transports.
    Replay(Mutex<Option<crate::transport::ReplayHandle>>),
}

impl Shutdown {
    /// Graceful close. Idempotent: a second call is a no-op.
    pub async fn shutdown(&self) -> Result<(), ConnectError> {
        match self {
            Shutdown::Stdio(slot) => {
                // release the guard before awaiting shutdown so concurrent shutdowns are not serialized through the slot Mutex.
                let handle = slot.lock().await.take();
                if let Some(handle) = handle {
                    handle
                        .shutdown()
                        .await
                        .map_err(|e| ConnectError::Shutdown(e.to_string()))?;
                }
            }
            Shutdown::StreamableHttp(slot) => {
                let handle = slot.lock().await.take();
                if let Some(handle) = handle {
                    handle.shutdown().await;
                }
            }
            Shutdown::LegacySse(slot) => {
                let handle = slot.lock().await.take();
                if let Some(handle) = handle {
                    handle.shutdown().await;
                }
            }
            Shutdown::Replay(slot) => {
                let handle = slot.lock().await.take();
                if let Some(handle) = handle {
                    handle.shutdown().await;
                }
            }
        }
        Ok(())
    }
}

/// Errors raised by [`connect_server`] or [`Shutdown::shutdown`].
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ConnectError {
    /// Stdio transport failed to spawn or initialize.
    #[error("stdio transport: {0}")]
    Stdio(#[from] StdioError),
    /// Streamable HTTP transport failed to connect.
    #[error("streamable-http transport: {0}")]
    StreamableHttp(#[from] StreamableHttpError),
    /// Legacy SSE transport failed to connect.
    #[error("legacy-sse transport: {0}")]
    LegacySse(#[from] LegacySseError),
    /// Cassette file could not be read or parsed for replay.
    #[error("cassette replay: {0}")]
    Cassette(String),
    /// OAuth client-credentials token acquisition failed before connect.
    #[error("oauth client-credentials: {0}")]
    OAuth(String),
    /// MCP `initialize` handshake failed.
    #[error("initialize handshake failed: {0}")]
    Initialize(#[from] ProtocolError),
    /// Graceful shutdown failed.
    #[error("shutdown failed: {0}")]
    Shutdown(String),
}

/// Connect to one server, run the MCP `initialize` handshake, and
/// return the live [`Client`] plus the shutdown handle.
///
/// `client_name` and `client_version` flow into the `initialize` call
/// so the server can log who is talking to it.
pub async fn connect_server(
    _name: &str,
    spec: &ServerSpec,
    client_name: &str,
    client_version: &str,
    proxy: &crate::network::ProxyConfig,
    extra_stdio_args: &[String],
) -> Result<LiveServer, ConnectError> {
    let transport_kind = match spec {
        ServerSpec::Stdio { .. } => "stdio",
        ServerSpec::StreamableHttp { .. } => "streamable-http",
        ServerSpec::LegacySse { .. } => "legacy-sse",
        ServerSpec::Cassette { .. } => "cassette-replay",
    };
    tracing::info!(
        target: "mcptest_core::connector",
        server = %_name,
        transport = transport_kind,
        "connect start"
    );
    match spec {
        ServerSpec::Stdio { command, args, env } => {
            let mut env_map = HashMap::new();
            for (k, v) in env {
                env_map.insert(k.clone(), v.clone());
            }
            let mut combined_args = args.clone();
            combined_args.extend(extra_stdio_args.iter().cloned());
            let cfg = StdioConfig {
                command: command.clone(),
                args: combined_args,
                env: env_map,
                ..StdioConfig::default()
            };
            connect_stdio(_name, cfg, client_name, client_version).await
        }
        ServerSpec::StreamableHttp {
            url,
            headers,
            oauth,
            stateless,
        } => {
            connect_streamable_http(
                url,
                headers,
                oauth,
                client_name,
                client_version,
                proxy,
                *stateless,
            )
            .await
        }
        ServerSpec::LegacySse {
            url,
            headers,
            oauth,
        } => {
            let mut header_map = HashMap::new();
            for (k, v) in headers {
                header_map.insert(k.clone(), v.clone());
            }
            resolve_url_auth(url, oauth, &mut header_map).await?;
            let cfg = LegacySseConfig {
                sse_url: url.clone(),
                headers: header_map,
                proxy: proxy.clone(),
                ..LegacySseConfig::default()
            };
            let (transport, handle) = LegacySse::connect(cfg).await?;
            let header_capture = Some(handle.header_capture());
            let client = build_client(transport, client_name, client_version).await;
            initialize_client(&client).await?;
            tracing::info!(
                target: "mcptest_core::connector",
                server = %_name,
                transport = "legacy-sse",
                "connect success"
            );
            Ok(LiveServer {
                client,
                shutdown: Shutdown::LegacySse(Mutex::new(Some(handle))),
                header_capture,
            })
        }
        ServerSpec::Cassette { path, .. } => {
            connect_cassette(path, client_name, client_version).await
        }
    }
}

/// Connect a [`ServerSpec::Stdio`] server: spawn the child, run the
/// handshake, and diagnose a handshake failure as a child exit when the
/// process is already gone (WOR-1326).
///
/// Why the diagnosis lives here: the transport sees the exit (stderr
/// EOF, reapable status) and the protocol layer sees the failed
/// request, but only the connector holds both halves. Mapping the
/// failure to [`StdioError::StartupExit`] turns "request timed out" or
/// "transport closed" into "the server exited with status X, here is
/// its stderr", which is what the operator can act on.
async fn connect_stdio(
    name: &str,
    cfg: StdioConfig,
    client_name: &str,
    client_version: &str,
) -> Result<LiveServer, ConnectError> {
    let (transport, mut handle) = StdioTransport::spawn(cfg)?;
    let client = build_client(transport, client_name, client_version).await;
    if let Err(err) = initialize_client(&client).await {
        // Best-effort teardown of the half-open session; the handshake
        // already failed, so a close error adds nothing.
        let _ = client.close().await;
        if let Some(status) = handle.exit_status_within(STARTUP_EXIT_GRACE).await {
            return Err(ConnectError::Stdio(StdioError::StartupExit {
                status,
                stderr_tail: handle.stderr_tail(),
            }));
        }
        return Err(err.into());
    }
    tracing::info!(
        target: "mcptest_core::connector",
        server = %name,
        transport = "stdio",
        "connect success"
    );
    Ok(LiveServer {
        client,
        shutdown: Shutdown::Stdio(Mutex::new(Some(handle))),
        header_capture: None,
    })
}

/// Connect a [`ServerSpec::StreamableHttp`] server: resolve auth headers
/// (explicit, OAuth client-credentials, or cached login token), open the
/// transport, and run the handshake. Pulled out of [`connect_server`] to keep
/// that function under the line cap.
async fn connect_streamable_http(
    url: &str,
    headers: &std::collections::BTreeMap<String, String>,
    oauth: &Option<OAuthClientCredentials>,
    client_name: &str,
    client_version: &str,
    proxy: &crate::network::ProxyConfig,
    stateless: bool,
) -> Result<LiveServer, ConnectError> {
    let mut header_map = HashMap::new();
    for (k, v) in headers {
        header_map.insert(k.clone(), v.clone());
    }
    resolve_url_auth(url, oauth, &mut header_map).await?;
    // The stateless 2026-07-28 shape carries the client's protocol version,
    // info, and capabilities in a per-request `_meta` block instead of the
    // one-time `initialize` handshake (WOR-1383).
    let caps = client_capabilities(client_name, client_version);
    let (mode, stateless_meta) = if stateless {
        (
            crate::transport::TransportMode::Stateless,
            Some(crate::protocol::build_request_meta(
                ProtocolVersion::V2026_07_28,
                &caps,
            )),
        )
    } else {
        (crate::transport::TransportMode::Session, None)
    };
    let cfg = StreamableHttpConfig {
        url: url.to_string(),
        headers: header_map,
        proxy: proxy.clone(),
        mode,
        stateless_meta,
        ..StreamableHttpConfig::default()
    };
    let (transport, handle) = StreamableHttp::connect(cfg)?;
    let header_capture = Some(handle.header_capture());
    let client = Client::builder(transport)
        .with_client_capabilities(caps)
        .with_request_timeout(DEFAULT_REQUEST_TIMEOUT)
        .build();
    // Stateless targets skip the handshake and fetch capabilities on demand
    // via `server/discover`; session targets run the `initialize` handshake.
    if stateless {
        let outcome = client.discover().await?;
        tracing::debug!(
            target: "mcptest_core::connector",
            negotiated_version = ?outcome.negotiated_version,
            "stateless server/discover complete"
        );
    } else {
        initialize_client(&client).await?;
    }
    tracing::info!(
        target: "mcptest_core::connector",
        transport = if stateless { "stateless-http" } else { "streamable-http" },
        "connect success"
    );
    Ok(LiveServer {
        client,
        shutdown: Shutdown::StreamableHttp(Mutex::new(Some(handle))),
        header_capture,
    })
}

/// Connect a [`ServerSpec::Cassette`] server: load the cassette file and serve
/// it through a [`ReplayTransport`](crate::transport::ReplayTransport). Pulled
/// out of [`connect_server`] to keep that function under the line cap.
async fn connect_cassette(
    path: &str,
    client_name: &str,
    client_version: &str,
) -> Result<LiveServer, ConnectError> {
    let raw = std::fs::read_to_string(path)
        .map_err(|e| ConnectError::Cassette(format!("read {path}: {e}")))?;
    let cassette: crate::cassette::Cassette = serde_json::from_str(&raw)
        .map_err(|e| ConnectError::Cassette(format!("parse {path}: {e}")))?;
    let (transport, handle) = crate::transport::ReplayTransport::spawn(
        cassette,
        crate::cassette::ReplayStrategy::Semantic,
    );
    let client = build_client(transport, client_name, client_version).await;
    initialize_client(&client).await?;
    tracing::info!(
        target: "mcptest_core::connector",
        transport = "cassette-replay",
        "connect success"
    );
    Ok(LiveServer {
        client,
        shutdown: Shutdown::Replay(Mutex::new(Some(handle))),
        header_capture: None,
    })
}

/// Resolve the token cache the connector reads. Honors `MCPTEST_AUTH_DIR` so
/// the connector and `mcptest login` share one root; otherwise the user
/// default `~/.mcptest/auth/`.
fn resolve_token_cache() -> Option<TokenCache> {
    match std::env::var_os("MCPTEST_AUTH_DIR") {
        Some(dir) => Some(TokenCache::with_root(PathBuf::from(dir))),
        None => TokenCache::user_default().ok(),
    }
}

/// Look up a cached, unexpired OAuth access token for `url` and return it as a
/// `Bearer` header value. Returns `None` when there is no cached token, the
/// cache cannot be read, or the token has expired (a WARN points the operator
/// at `mcptest login`).
fn bearer_if_valid(cache: &TokenCache, url: &str, now: u64) -> Option<String> {
    let token = cache.read(url).ok()??;
    if token.is_expired_at(now) {
        return None;
    }
    Some(format!("Bearer {}", token.access_token))
}

/// A cached `mcptest login` bearer for `url`. Returns the token when valid,
/// refreshes it in place when it has expired but carries the metadata to do so
///, and `None` when there is no usable token (with a re-login hint).
async fn cached_login_bearer(cache: &TokenCache, url: &str) -> Option<String> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
    if let Some(bearer) = bearer_if_valid(cache, url, now) {
        return Some(bearer);
    }
    let token = cache.read(url).ok()??;
    refresh_cached_token(cache, url, &token).await
}

/// Refresh an expired cached token in place using its `refresh_token` and the
/// `token_endpoint`/`client_id` stamped at login. Persists the refreshed token
/// so later runs can refresh again. `None` (with a warning) when the token is
/// not refreshable or the refresh call fails.
async fn refresh_cached_token(cache: &TokenCache, url: &str, token: &Token) -> Option<String> {
    let (refresh_token, endpoint, client_id) = match (
        token.refresh_token.clone(),
        token.token_endpoint.clone(),
        token.client_id.clone(),
    ) {
        (Some(rt), Some(ep), Some(cid)) => (rt, ep, cid),
        _ => {
            tracing::warn!(
                target: "mcptest_core::connector",
                url = %url,
                "cached OAuth token has expired and cannot be refreshed; run `mcptest login`"
            );
            return None;
        }
    };
    let cfg = ReqwestRefresherConfig {
        token_endpoint: endpoint.clone(),
        client_id: client_id.clone(),
        client_secret: None,
        ..Default::default()
    };
    let prev_refresh = refresh_token.clone();
    // Build and run the blocking refresher entirely off the async runtime: a
    // reqwest::blocking client owns its own runtime, which cannot be created or
    // dropped inside an async context.
    let refreshed = tokio::task::spawn_blocking(move || {
        let refresher = ReqwestRefresher::new(cfg)?;
        refresher.refresh(&refresh_token)
    })
    .await
    .ok()?;
    let mut new_token = match refreshed {
        Ok(t) => t,
        Err(e) => {
            tracing::warn!(
                target: "mcptest_core::connector",
                url = %url,
                "cached OAuth token refresh failed ({e}); run `mcptest login`"
            );
            return None;
        }
    };
    // Re-stamp the refresh metadata, and keep the old refresh token if the IdP
    // did not rotate it, so the next run can refresh again.
    new_token.token_endpoint = Some(endpoint);
    new_token.client_id = Some(client_id);
    if new_token.refresh_token.is_none() {
        new_token.refresh_token = Some(prev_refresh);
    }
    let bearer = format!("Bearer {}", new_token.access_token);
    if let Err(e) = cache.write(url, &new_token) {
        // WOR-1507: a failed persist (read-only cache dir, full disk) was
        // silently dropped, so every run re-refreshed and, with an IdP that
        // rotates single-use refresh tokens, the next run failed auth with no
        // clue why. Warn, naming the failure, instead of swallowing it.
        tracing::warn!(
            target: "mcptest_core::connector",
            url = %url,
            error = %e,
            "refreshed the OAuth token but could not persist it to the auth cache; the next run will re-refresh"
        );
    }
    tracing::info!(
        target: "mcptest_core::connector",
        url = %url,
        "refreshed an expired cached OAuth token"
    );
    Some(bearer)
}

/// Decide the Authorization header for a URL server. Precedence: an explicit
/// header already set by `auth.bearer_token_env` wins; then an `auth.oauth`
/// client-credentials exchange; then a cached `mcptest login` token.
async fn resolve_url_auth(
    url: &str,
    oauth: &Option<OAuthClientCredentials>,
    headers: &mut HashMap<String, String>,
) -> Result<(), ConnectError> {
    if headers
        .keys()
        .any(|k| k.eq_ignore_ascii_case("authorization"))
    {
        return Ok(());
    }
    if let Some(cfg) = oauth {
        let bearer = client_credentials_bearer(cfg).await?;
        headers.insert("Authorization".to_string(), bearer);
    } else if let Some(cache) = resolve_token_cache() {
        if let Some(bearer) = cached_login_bearer(&cache, url).await {
            headers.insert("Authorization".to_string(), bearer);
        }
    }
    Ok(())
}

/// Run the client-credentials grant and return the `Bearer <token>` value. The
/// secret is read from the configured env var at connect time (never stored in
/// the suite); a missing env var or a failed exchange is a [`ConnectError`].
async fn client_credentials_bearer(cfg: &OAuthClientCredentials) -> Result<String, ConnectError> {
    let secret = std::env::var(&cfg.client_secret_env).map_err(|_| {
        ConnectError::OAuth(format!(
            "client secret env var `{}` is not set",
            cfg.client_secret_env
        ))
    })?;
    let req = crate::auth::ClientCredentialsRequest {
        token_endpoint: cfg.token_endpoint.clone(),
        client_id: cfg.client_id.clone(),
        client_secret: secret,
        scope: cfg.scope.clone(),
    };
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    // The exchange uses a blocking reqwest client; run it off the async runtime.
    let token =
        tokio::task::spawn_blocking(move || crate::auth::exchange_client_credentials(&req, now))
            .await
            .map_err(|e| ConnectError::OAuth(format!("client-credentials task failed: {e}")))?
            .map_err(|e| ConnectError::OAuth(format!("client-credentials exchange failed: {e}")))?;
    Ok(format!("Bearer {}", token.access_token))
}

/// Build the client capability block advertised to the server (in the
/// `initialize` handshake, or in every stateless request's `_meta`).
fn client_capabilities(client_name: &str, client_version: &str) -> ClientCapabilities {
    ClientCapabilities {
        name: client_name.to_string(),
        version: client_version.to_string(),
        capabilities: serde_json::json!({}),
    }
}

async fn build_client(
    transport: crate::protocol::Transport,
    client_name: &str,
    client_version: &str,
) -> Client {
    Client::builder(transport)
        .with_client_capabilities(client_capabilities(client_name, client_version))
        .with_request_timeout(DEFAULT_REQUEST_TIMEOUT)
        .build()
}

async fn initialize_client(client: &Client) -> Result<(), ProtocolError> {
    // Request the most recent supported protocol version; the server
    // negotiates down if it predates it.
    let info = client.initialize(ProtocolVersion::V2025_03_26).await?;
    tracing::debug!(
        target: "mcptest_core::connector",
        negotiated_version = ?info,
        "initialize handshake complete"
    );
    Ok(())
}

/// Pool of connected servers, keyed by suite-local name.
///
/// The executor builds one of these once per run and looks up servers
/// by the name referenced from each test's `server:` field.
pub struct ServerPool {
    inner: Arc<Mutex<HashMap<String, Arc<LiveServer>>>>,
}

impl ServerPool {
    /// Build a new empty pool.
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Insert a connected server under the supplied name.
    pub async fn insert(&self, name: String, server: LiveServer) {
        let server = Arc::new(server);
        let mut map = self.inner.lock().await;
        map.insert(name, server);
    }

    /// Fetch a connected server by name. Returns `None` when the test
    /// references a server the loader did not declare.
    pub async fn get(&self, name: &str) -> Option<Arc<LiveServer>> {
        let map = self.inner.lock().await;
        map.get(name).cloned()
    }

    /// Remove a single server from the pool and shut it down, leaving
    /// every other entry intact. Used by the `restart_policy`
    /// integration so a `per_file` or `per_test` policy can clean up
    /// just one server between test boundaries without disturbing the
    /// rest of the suite.
    ///
    /// Returns `Ok(true)` when the server was present and the shutdown
    /// succeeded; `Ok(false)` when no entry was found under `name`;
    /// `Err(_)` when the entry was present but the shutdown failed.
    /// A subsequent `insert` is required to reconnect the server; the
    /// caller decides whether to re-spawn or fall through.
    pub async fn shutdown_one(&self, name: &str) -> Result<bool, ConnectError> {
        let removed = {
            let mut map = self.inner.lock().await;
            map.remove(name)
        };
        let Some(entry) = removed else {
            return Ok(false);
        };
        entry.shutdown.shutdown().await?;
        Ok(true)
    }

    /// Snapshot the protocol violations recorded on every pooled
    /// connection, sorted by server name (WOR-1311).
    ///
    /// Why: the run report attaches violations per run, not per test,
    /// and only the pool knows which client belongs to which declared
    /// server. Call this before `shutdown_all`, while the clients are
    /// still alive.
    pub async fn protocol_violations(
        &self,
    ) -> Vec<(String, crate::protocol::ProtocolViolationSummary)> {
        let entries: Vec<(String, Arc<LiveServer>)> = {
            let map = self.inner.lock().await;
            map.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
        };
        let mut out = Vec::with_capacity(entries.len());
        for (name, server) in entries {
            out.push((name, server.client.protocol_violations().await));
        }
        out.sort_by(|a, b| a.0.cmp(&b.0));
        out
    }

    /// Shut down every server in the pool. The runner calls this in
    /// `finalize` so subprocess servers exit cleanly.
    pub async fn shutdown_all(&self) -> Vec<ConnectError> {
        let entries: Vec<Arc<LiveServer>> = {
            let mut map = self.inner.lock().await;
            map.drain().map(|(_, v)| v).collect()
        };
        let mut errors = Vec::new();
        for entry in entries {
            // Each LiveServer is Arc'd into the executor; once the
            // executor drops its clones, this is the last reference
            // and Arc::try_unwrap succeeds. If it does not, the
            // shutdown still runs through the Mutex slot.
            if let Err(e) = entry.shutdown.shutdown().await {
                tracing::warn!(
                    target: "mcptest_core::connector",
                    error = %e,
                    "shutdown error"
                );
                errors.push(e);
            }
        }
        errors
    }
}

impl Default for ServerPool {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[tokio::test]
    async fn shutdown_on_empty_pool_returns_no_errors() {
        let pool = ServerPool::new();
        let errs = pool.shutdown_all().await;
        assert!(errs.is_empty());
    }

    #[tokio::test]
    async fn pool_get_returns_none_for_unknown_server() {
        let pool = ServerPool::new();
        assert!(pool.get("missing").await.is_none());
    }

    /// Minimal cassette: just the `initialize` handshake, enough to build
    /// a pool entry without any subprocess or network (WOR-1323).
    const HANDSHAKE_CASSETTE: &str = r#"{
      "version": "1",
      "exchanges": [
        {
          "request": { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {} },
          "response": { "jsonrpc": "2.0", "id": 1, "result": { "protocolVersion": "2025-03-26", "capabilities": {}, "serverInfo": { "name": "recorded", "version": "0.1" } } }
        }
      ]
    }"#;

    async fn cassette_entry(dir: &std::path::Path) -> LiveServer {
        let path = dir.join("cassette.json");
        std::fs::write(&path, HANDSHAKE_CASSETTE).unwrap();
        connect_cassette(path.to_str().unwrap(), "mcptest", "0.0.0")
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn insert_replaces_an_existing_entry_under_the_same_name() {
        // The restart-policy respawn (WOR-1323) relies on `insert` swapping
        // the entry the executor's per-test `get` resolves; a duplicate-key
        // refusal would silently keep the dead server.
        let dir = tempfile::tempdir().unwrap();
        let pool = ServerPool::new();
        pool.insert("srv".to_string(), cassette_entry(dir.path()).await)
            .await;
        let first = pool.get("srv").await.unwrap();
        pool.insert("srv".to_string(), cassette_entry(dir.path()).await)
            .await;
        let second = pool.get("srv").await.unwrap();
        assert!(
            !Arc::ptr_eq(&first, &second),
            "insert under an existing name must replace the entry"
        );
    }

    #[tokio::test]
    async fn shutdown_one_on_a_missing_name_reports_false_without_error() {
        let pool = ServerPool::new();
        let removed = pool.shutdown_one("ghost").await.unwrap();
        assert!(!removed, "a missing entry is Ok(false), not an error");
    }

    // Stdio connect tests are exercised end-to-end by the executor
    // integration tests; spawning a real subprocess here would
    // duplicate that coverage and slow the unit-test suite. The one
    // exception is the startup-exit diagnosis (WOR-1326), which only
    // the connector can produce.

    #[cfg(unix)]
    #[tokio::test]
    async fn handshake_against_an_exiting_command_fails_fast_with_status_and_stderr() {
        let cfg = StdioConfig {
            command: "sh".into(),
            args: vec![
                "-c".into(),
                "echo \"Error: Cannot find module '/nope.js'\" >&2; exit 7".into(),
            ],
            ..StdioConfig::default()
        };
        let started = std::time::Instant::now();
        let err = match connect_stdio("dying", cfg, "mcptest", "0.0.0").await {
            Ok(_) => panic!("child exits before answering initialize"),
            Err(err) => err,
        };
        assert!(
            started.elapsed() < Duration::from_secs(5),
            "diagnosis took {:?} instead of failing fast",
            started.elapsed()
        );
        match err {
            ConnectError::Stdio(StdioError::StartupExit {
                status,
                stderr_tail,
            }) => {
                assert_eq!(status.code(), Some(7));
                assert!(
                    stderr_tail.iter().any(|l| l.contains("Cannot find module")),
                    "stderr tail missing the cause: {stderr_tail:?}"
                );
            }
            other => panic!("expected StartupExit, got {other:?}"),
        }
    }

    fn token(access: &str, expires_at_unix: u64) -> Token {
        Token {
            access_token: access.to_string(),
            refresh_token: None,
            expires_at_unix,
            token_type: "Bearer".to_string(),
            scope: None,
            last_refreshed: "2026-05-26T00:00:00Z".to_string(),
            token_endpoint: None,
            client_id: None,
        }
    }

    fn far_future() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs()
            + 3600
    }

    #[test]
    fn bearer_if_valid_returns_an_unexpired_token() {
        let dir = tempfile::tempdir().unwrap();
        let cache = TokenCache::with_root(dir.path());
        let url = "https://mcp.example.com/secure/mcp";
        cache
            .write(url, &token("cached-abc", far_future()))
            .unwrap();
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        assert_eq!(
            bearer_if_valid(&cache, url, now).as_deref(),
            Some("Bearer cached-abc")
        );
    }

    #[test]
    fn bearer_if_valid_is_none_for_an_expired_token() {
        let dir = tempfile::tempdir().unwrap();
        let cache = TokenCache::with_root(dir.path());
        let url = "https://mcp.example.com/secure/mcp";
        cache.write(url, &token("stale", 1)).unwrap(); // expired in 1970
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        assert!(bearer_if_valid(&cache, url, now).is_none());
    }

    #[test]
    fn bearer_if_valid_is_none_with_no_cached_token() {
        let dir = tempfile::tempdir().unwrap();
        let cache = TokenCache::with_root(dir.path());
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        assert!(bearer_if_valid(&cache, "https://unknown.example.com/mcp", now).is_none());
    }

    #[tokio::test]
    async fn refresh_cached_token_refreshes_an_expired_token_in_place() {
        use wiremock::matchers::{body_string_contains, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .and(body_string_contains("grant_type=refresh_token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "refreshed-AT",
                "expires_in": 3600,
                "token_type": "Bearer"
            })))
            .mount(&server)
            .await;

        let dir = tempfile::tempdir().unwrap();
        let cache = TokenCache::with_root(dir.path());
        let url = "https://mcp.example.com/secure/mcp";
        // An expired token that carries the metadata to refresh itself.
        let mut expired = token("old-AT", 1);
        expired.refresh_token = Some("RT.original".to_string());
        expired.token_endpoint = Some(format!("{}/token", server.uri()));
        expired.client_id = Some("svc".to_string());
        cache.write(url, &expired).unwrap();

        let bearer = cached_login_bearer(&cache, url).await;
        assert_eq!(bearer.as_deref(), Some("Bearer refreshed-AT"));
        // The refreshed token is persisted with its refresh metadata intact, so
        // the next run can refresh again; the IdP did not rotate the refresh
        // token here, so the original is kept.
        let stored = cache.read(url).unwrap().unwrap();
        assert_eq!(stored.access_token, "refreshed-AT");
        assert_eq!(stored.refresh_token.as_deref(), Some("RT.original"));
        assert!(stored.token_endpoint.is_some() && stored.client_id.is_some());
    }

    #[tokio::test]
    async fn cached_login_bearer_is_none_for_an_unrefreshable_expired_token() {
        let dir = tempfile::tempdir().unwrap();
        let cache = TokenCache::with_root(dir.path());
        let url = "https://mcp.example.com/secure/mcp";
        cache.write(url, &token("stale", 1)).unwrap(); // expired, no refresh metadata
        assert!(cached_login_bearer(&cache, url).await.is_none());
    }

    #[tokio::test]
    async fn stateless_server_spec_connects_via_discover_not_initialize() {
        // A stateless `ServerSpec` must drive the 2026-07-28 transport: fetch
        // capabilities via `server/discover` (no `initialize` handshake), carry
        // `_meta` on the request, and never send a session id (WOR-1383). The
        // mock mounts ONLY `server/discover`, so a connect that still called
        // `initialize` would fail to match and error out.
        use crate::transport::{STREAMABLE_HTTP_METHOD_HEADER, STREAMABLE_HTTP_SESSION_HEADER};
        use wiremock::matchers::{body_string_contains, header, method, path};
        use wiremock::{Mock, MockServer, Request, ResponseTemplate};

        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/mcp"))
            .and(header(STREAMABLE_HTTP_METHOD_HEADER, "server/discover"))
            .and(body_string_contains("_meta"))
            .respond_with(|req: &Request| {
                let body: serde_json::Value =
                    serde_json::from_slice(&req.body).unwrap_or(serde_json::Value::Null);
                let id = body.get("id").cloned().unwrap_or(serde_json::Value::Null);
                ResponseTemplate::new(200)
                    .insert_header("content-type", "application/json")
                    .set_body_json(serde_json::json!({
                        "jsonrpc": "2.0",
                        "id": id,
                        "result": {
                            "protocolVersion": "2026-07-28",
                            "serverInfo": { "name": "stateless", "version": "1.0.0" },
                            "capabilities": { "tools": { "listChanged": false } }
                        }
                    }))
            })
            .mount(&server)
            .await;

        let spec = ServerSpec::StreamableHttp {
            url: format!("{}/mcp", server.uri()),
            headers: Default::default(),
            oauth: None,
            stateless: true,
        };
        let _live = connect_server(
            "api",
            &spec,
            "mcptest",
            "1.0.0",
            &crate::network::ProxyConfig::default(),
            &[],
        )
        .await
        .expect("stateless connect should succeed via server/discover");

        let requests = server.received_requests().await.unwrap();
        let header_str = |req: &Request, name: &str| -> Option<String> {
            req.headers
                .get(name)
                .and_then(|v| v.to_str().ok())
                .map(str::to_string)
        };
        assert!(
            requests
                .iter()
                .all(|r| header_str(r, STREAMABLE_HTTP_METHOD_HEADER).as_deref()
                    != Some("initialize")),
            "a stateless connect must not run the initialize handshake"
        );
        let discover = requests
            .iter()
            .find(|r| {
                header_str(r, STREAMABLE_HTTP_METHOD_HEADER).as_deref() == Some("server/discover")
            })
            .expect("a server/discover request was recorded");
        assert!(
            discover
                .headers
                .get(STREAMABLE_HTTP_SESSION_HEADER)
                .is_none(),
            "stateless transport must not send a session id"
        );
        let body: serde_json::Value = serde_json::from_slice(&discover.body).unwrap();
        assert_eq!(body["params"]["_meta"]["protocolVersion"], "2026-07-28");
        // `_live` drops at end of scope, aborting the transport tasks.
    }
}
