tower-mcp 0.25.0

//! Cohesive OAuth authorization-code state machine for MCP clients.
//!
//! [`OAuthAuthorizationFlow`] composes the lower-level discovery,
//! registration, PKCE, redirect, token, refresh, persistence, and scope
//! escalation pieces into one reusable client.

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

use async_trait::async_trait;
use base64::Engine;
use tokio::sync::{Mutex, RwLock, oneshot};

use super::oauth::{
    OAuthBearerChallenge, OAuthClientError, OAuthScopeEscalationHandler,
    OAuthScopeEscalationRequest, OAuthTokenEndpointAuthMethod, TokenProvider,
};
use super::oauth_authcode::{
    OAuthAuthorizationServerMetadata, OAuthClientRegistration, OAuthClientRegistrationMethod,
    OAuthClientRegistrationOptions, OAuthClientRegistrationStore, OAuthProtectedResourceMetadata,
    discovery_reason, preferred_registration_auth_method,
    require_secret_for_confidential_registration, validate_resource_identifier,
    validate_resource_override,
};

/// HTTP method used by an OAuth protocol request.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum OAuthHttpMethod {
    /// HTTP GET.
    Get,
    /// HTTP POST.
    Post,
}

/// Body of an OAuth protocol HTTP request.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum OAuthHttpBody {
    /// No request body.
    Empty,
    /// `application/x-www-form-urlencoded` fields.
    Form(Vec<(String, String)>),
    /// JSON request body.
    Json(serde_json::Value),
}

/// Transport-neutral OAuth protocol HTTP request.
#[derive(Debug, Clone)]
pub struct OAuthHttpRequest {
    /// HTTP method.
    pub method: OAuthHttpMethod,
    /// Absolute request URL.
    pub url: String,
    /// Request headers.
    pub headers: Vec<(String, String)>,
    /// Request body.
    pub body: OAuthHttpBody,
}

impl OAuthHttpRequest {
    fn get(url: impl Into<String>) -> Self {
        Self {
            method: OAuthHttpMethod::Get,
            url: url.into(),
            headers: Vec::new(),
            body: OAuthHttpBody::Empty,
        }
    }

    fn post_form(url: impl Into<String>, fields: Vec<(String, String)>) -> Self {
        Self {
            method: OAuthHttpMethod::Post,
            url: url.into(),
            headers: Vec::new(),
            body: OAuthHttpBody::Form(fields),
        }
    }

    fn post_json(url: impl Into<String>, value: serde_json::Value) -> Self {
        Self {
            method: OAuthHttpMethod::Post,
            url: url.into(),
            headers: Vec::new(),
            body: OAuthHttpBody::Json(value),
        }
    }

    fn basic_auth(mut self, client_id: &str, client_secret: &str) -> Self {
        // RFC 6749 section 2.3.1 applies application/x-www-form-urlencoded
        // encoding to both credentials before constructing HTTP Basic auth.
        let client_id = urlencoding::encode(client_id);
        let client_secret = urlencoding::encode(client_secret);
        let encoded = base64::engine::general_purpose::STANDARD
            .encode(format!("{client_id}:{client_secret}"));
        self.headers
            .push(("authorization".to_string(), format!("Basic {encoded}")));
        self
    }
}

/// Transport-neutral OAuth protocol HTTP response.
#[derive(Debug, Clone)]
pub struct OAuthHttpResponse {
    /// Numeric HTTP status code.
    pub status: u16,
    /// Response headers. Repeated fields are retained as separate entries.
    pub headers: Vec<(String, String)>,
    /// Complete response body.
    pub body: Vec<u8>,
}

impl OAuthHttpResponse {
    /// Return whether the response status is in the 2xx range.
    pub fn is_success(&self) -> bool {
        (200..300).contains(&self.status)
    }

    /// Iterate over values for a case-insensitive header name.
    pub fn header_values<'a>(&'a self, name: &'a str) -> impl Iterator<Item = &'a str> + 'a {
        self.headers
            .iter()
            .filter(move |(candidate, _)| candidate.eq_ignore_ascii_case(name))
            .map(|(_, value)| value.as_str())
    }

    fn json<T: serde::de::DeserializeOwned>(&self) -> Result<T, OAuthClientError> {
        serde_json::from_slice(&self.body)
            .map_err(|error| OAuthClientError::InvalidResponse(error.to_string()))
    }

    fn error_body(&self) -> String {
        String::from_utf8_lossy(&self.body)
            .chars()
            .take(1024)
            .collect()
    }
}

/// Application-supplied HTTP abstraction for OAuth protocol traffic.
///
/// Implementations control proxying, TLS, telemetry, retries, and test
/// behavior without requiring the authorization state machine to depend on a
/// particular HTTP stack.
#[async_trait]
pub trait OAuthHttpClient: Send + Sync + 'static {
    /// Execute one OAuth protocol request.
    async fn execute(
        &self,
        request: OAuthHttpRequest,
    ) -> Result<OAuthHttpResponse, OAuthClientError>;
}

/// [`OAuthHttpClient`] backed by reqwest.
#[derive(Clone)]
pub struct ReqwestOAuthHttpClient {
    client: reqwest::Client,
}

impl fmt::Debug for ReqwestOAuthHttpClient {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ReqwestOAuthHttpClient")
            .finish_non_exhaustive()
    }
}

impl ReqwestOAuthHttpClient {
    /// Wrap an application-configured reqwest client.
    pub fn new(client: reqwest::Client) -> Self {
        Self { client }
    }

    /// Create a client that does not automatically follow redirects.
    pub fn without_redirects() -> Result<Self, OAuthClientError> {
        reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .map(Self::new)
            .map_err(|error| OAuthClientError::Http(error.to_string()))
    }
}

#[async_trait]
impl OAuthHttpClient for ReqwestOAuthHttpClient {
    async fn execute(
        &self,
        request: OAuthHttpRequest,
    ) -> Result<OAuthHttpResponse, OAuthClientError> {
        let method = match request.method {
            OAuthHttpMethod::Get => reqwest::Method::GET,
            OAuthHttpMethod::Post => reqwest::Method::POST,
        };
        let mut builder = self.client.request(method, &request.url);
        for (name, value) in request.headers {
            builder = builder.header(name, value);
        }
        builder = match request.body {
            OAuthHttpBody::Empty => builder,
            OAuthHttpBody::Form(fields) => builder.form(&fields),
            OAuthHttpBody::Json(value) => builder.json(&value),
        };
        let response = builder
            .send()
            .await
            .map_err(|error| OAuthClientError::Http(error.to_string()))?;
        let status = response.status().as_u16();
        let headers = response
            .headers()
            .iter()
            .filter_map(|(name, value)| {
                value
                    .to_str()
                    .ok()
                    .map(|value| (name.as_str().to_string(), value.to_string()))
            })
            .collect();
        let body = response
            .bytes()
            .await
            .map_err(|error| OAuthClientError::Http(error.to_string()))?
            .to_vec();
        Ok(OAuthHttpResponse {
            status,
            headers,
            body,
        })
    }
}

/// Binding key that prevents token reuse across resources, issuers, or clients.
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct OAuthTokenBinding {
    /// Canonical MCP protected-resource identifier.
    pub resource: String,
    /// Exact authorization-server issuer.
    pub issuer: String,
    /// OAuth client identifier.
    pub client_id: String,
}

/// Persistable OAuth authorization-code token set.
///
/// This value contains bearer credentials and must be protected as a secret.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct OAuthStoredToken {
    /// Access token.
    pub access_token: String,
    /// Refresh token, when issued.
    pub refresh_token: Option<String>,
    /// Expiration time as Unix seconds.
    ///
    /// This is [`u64::MAX`] when the authorization server did not advertise
    /// an `expires_in` lifetime.
    pub expires_at: u64,
    /// Scopes represented by this token.
    pub scopes: Vec<String>,
}

impl fmt::Debug for OAuthStoredToken {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OAuthStoredToken")
            .field("access_token", &"[REDACTED]")
            .field(
                "refresh_token",
                &self.refresh_token.as_ref().map(|_| "[REDACTED]"),
            )
            .field("expires_at", &self.expires_at)
            .field("scopes", &self.scopes)
            .finish()
    }
}

/// Persistent storage for resource/issuer/client-bound OAuth tokens.
#[async_trait]
pub trait OAuthTokenStore: Send + Sync {
    /// Load a token set for an exact binding.
    async fn load(
        &self,
        binding: &OAuthTokenBinding,
    ) -> Result<Option<OAuthStoredToken>, OAuthClientError>;

    /// Save a token set for an exact binding.
    async fn save(
        &self,
        binding: &OAuthTokenBinding,
        token: &OAuthStoredToken,
    ) -> Result<(), OAuthClientError>;

    /// Remove a token set for an exact binding.
    async fn remove(&self, binding: &OAuthTokenBinding) -> Result<(), OAuthClientError>;
}

/// Process-local OAuth token store.
#[derive(Clone, Default)]
pub struct MemoryOAuthTokenStore {
    tokens: Arc<RwLock<HashMap<OAuthTokenBinding, OAuthStoredToken>>>,
}

impl fmt::Debug for MemoryOAuthTokenStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("MemoryOAuthTokenStore")
            .finish_non_exhaustive()
    }
}

impl MemoryOAuthTokenStore {
    /// Create an empty process-local token store.
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait]
impl OAuthTokenStore for MemoryOAuthTokenStore {
    async fn load(
        &self,
        binding: &OAuthTokenBinding,
    ) -> Result<Option<OAuthStoredToken>, OAuthClientError> {
        Ok(self.tokens.read().await.get(binding).cloned())
    }

    async fn save(
        &self,
        binding: &OAuthTokenBinding,
        token: &OAuthStoredToken,
    ) -> Result<(), OAuthClientError> {
        self.tokens
            .write()
            .await
            .insert(binding.clone(), token.clone());
        Ok(())
    }

    async fn remove(&self, binding: &OAuthTokenBinding) -> Result<(), OAuthClientError> {
        self.tokens.write().await.remove(binding);
        Ok(())
    }
}

/// Persistable PKCE and CSRF state for an in-progress authorization.
///
/// This contains a PKCE verifier and possibly a client secret. Store it with
/// the same protections as OAuth credentials.
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct OAuthPendingAuthorizationState {
    /// CSRF state key.
    pub state: String,
    /// PKCE verifier.
    pub code_verifier: String,
    /// Exact redirect URI.
    pub redirect_uri: String,
    /// Canonical resource identifier.
    pub resource: String,
    /// Exact authorization-server issuer.
    pub issuer: String,
    /// Token endpoint selected from validated metadata.
    pub token_endpoint: String,
    /// Resolved client registration.
    pub registration: OAuthClientRegistration,
    /// Token endpoint authentication method.
    pub token_endpoint_auth_method: OAuthTokenEndpointAuthMethod,
    /// Requested scopes.
    pub scopes: Vec<String>,
    /// Whether an authorization-response `iss` parameter is required.
    pub iss_required: bool,
}

impl fmt::Debug for OAuthPendingAuthorizationState {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OAuthPendingAuthorizationState")
            .field("state", &"[REDACTED]")
            .field("code_verifier", &"[REDACTED]")
            .field("redirect_uri", &self.redirect_uri)
            .field("resource", &self.resource)
            .field("issuer", &self.issuer)
            .field("token_endpoint", &self.token_endpoint)
            .field("registration", &self.registration)
            .field(
                "token_endpoint_auth_method",
                &self.token_endpoint_auth_method,
            )
            .field("scopes", &self.scopes)
            .field("iss_required", &self.iss_required)
            .finish()
    }
}

/// Persistent storage for PKCE and CSRF state.
#[async_trait]
pub trait OAuthAuthorizationStateStore: Send + Sync {
    /// Load pending authorization state by the exact CSRF state key.
    async fn load(
        &self,
        state: &str,
    ) -> Result<Option<OAuthPendingAuthorizationState>, OAuthClientError>;

    /// Save pending authorization state.
    async fn save(&self, state: &OAuthPendingAuthorizationState) -> Result<(), OAuthClientError>;

    /// Remove consumed or abandoned state.
    async fn remove(&self, state: &str) -> Result<(), OAuthClientError>;
}

/// Process-local PKCE and CSRF state store.
#[derive(Clone, Default)]
pub struct MemoryOAuthAuthorizationStateStore {
    states: Arc<RwLock<HashMap<String, OAuthPendingAuthorizationState>>>,
}

impl fmt::Debug for MemoryOAuthAuthorizationStateStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("MemoryOAuthAuthorizationStateStore")
            .finish_non_exhaustive()
    }
}

impl MemoryOAuthAuthorizationStateStore {
    /// Create an empty process-local state store.
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait]
impl OAuthAuthorizationStateStore for MemoryOAuthAuthorizationStateStore {
    async fn load(
        &self,
        state: &str,
    ) -> Result<Option<OAuthPendingAuthorizationState>, OAuthClientError> {
        Ok(self.states.read().await.get(state).cloned())
    }

    async fn save(&self, state: &OAuthPendingAuthorizationState) -> Result<(), OAuthClientError> {
        self.states
            .write()
            .await
            .insert(state.state.clone(), state.clone());
        Ok(())
    }

    async fn remove(&self, state: &str) -> Result<(), OAuthClientError> {
        self.states.write().await.remove(state);
        Ok(())
    }
}

/// Redirect handling policy for an authorization flow.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum OAuthRedirectPolicy {
    /// Host a one-shot callback on the loopback interface.
    Loopback {
        /// Requested port, or `None` for an ephemeral port.
        port: Option<u16>,
        /// Callback path, beginning with `/`.
        callback_path: String,
    },
    /// Use an application-owned redirect URI and complete the flow by passing
    /// the returned callback URL to [`OAuthAuthorizationFlow::complete_callback_url`].
    Fixed {
        /// Exact registered redirect URI.
        redirect_uri: String,
    },
}

impl OAuthRedirectPolicy {
    /// Use an ephemeral loopback port and `/callback`.
    pub fn loopback() -> Self {
        Self::Loopback {
            port: None,
            callback_path: "/callback".to_string(),
        }
    }

    /// Use a specific loopback port and callback path.
    pub fn loopback_at(port: u16, callback_path: impl Into<String>) -> Self {
        Self::Loopback {
            port: Some(port),
            callback_path: callback_path.into(),
        }
    }

    /// Use an application-owned redirect URI.
    pub fn fixed(redirect_uri: impl Into<String>) -> Self {
        Self::Fixed {
            redirect_uri: redirect_uri.into(),
        }
    }
}

/// Authorization request presented to an application or browser integration.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OAuthAuthorizationRequest {
    /// URL that the user agent should open.
    pub authorization_url: String,
    /// Exact redirect URI registered for this attempt.
    pub redirect_uri: String,
    /// Canonical resource identifier.
    pub resource: String,
    /// Exact authorization-server issuer.
    pub issuer: String,
    /// Requested scopes.
    pub scopes: Vec<String>,
}

/// Result returned by an application authorization handler.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum OAuthAuthorizationAction {
    /// The handler presented the URL; wait for the configured loopback callback.
    AwaitLoopback,
    /// The handler captured and returned the complete callback URL.
    CallbackUrl(String),
}

/// Application hook that presents or automates an authorization redirect.
#[async_trait]
pub trait OAuthAuthorizationHandler: Send + Sync + 'static {
    /// Present the authorization request and choose how it completes.
    async fn authorize(
        &self,
        request: OAuthAuthorizationRequest,
    ) -> Result<OAuthAuthorizationAction, OAuthClientError>;
}

/// Input to a private-key JWT signer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OAuthClientAssertionRequest {
    /// OAuth client identifier (`iss` and `sub` in the assertion).
    pub client_id: String,
    /// Token endpoint URL (`aud` in the assertion).
    pub token_endpoint: String,
    /// Exact authorization-server issuer selected for the flow.
    pub authorization_server_issuer: String,
}

/// Application hook for `private_key_jwt` client authentication.
///
/// Implementations normally create a short-lived signed JWT containing `iss`,
/// `sub`, `aud`, `iat`, `exp`, and a unique `jti`.
#[async_trait]
pub trait OAuthClientAssertionSigner: Send + Sync + 'static {
    /// Create a client assertion for one token request.
    async fn sign_client_assertion(
        &self,
        request: OAuthClientAssertionRequest,
    ) -> Result<String, OAuthClientError>;
}

#[derive(Clone)]
struct ActiveToken {
    binding: OAuthTokenBinding,
    token: OAuthStoredToken,
    token_endpoint: String,
    registration: OAuthClientRegistration,
    auth_method: OAuthTokenEndpointAuthMethod,
}

struct FlowInner {
    resource_url: String,
    resource_override: Option<String>,
    registration_options: OAuthClientRegistrationOptions,
    pre_registered_client: Option<(String, Option<String>)>,
    registration_store: Arc<dyn OAuthClientRegistrationStore>,
    token_store: Arc<dyn OAuthTokenStore>,
    state_store: Arc<dyn OAuthAuthorizationStateStore>,
    http: Arc<dyn OAuthHttpClient>,
    redirect_policy: OAuthRedirectPolicy,
    preferred_issuer: Option<String>,
    refresh_buffer: Duration,
    authorization_handler: Option<Arc<dyn OAuthAuthorizationHandler>>,
    assertion_signer: Option<Arc<dyn OAuthClientAssertionSigner>>,
    current: RwLock<Option<ActiveToken>>,
    authorization_lock: Mutex<()>,
    refresh_lock: Mutex<()>,
}

/// Reusable OAuth authorization-code state machine and token provider.
///
/// Build this type with [`OAuthAuthorizationFlow::builder`], call
/// [`authorize`](Self::authorize) for a fully driven flow or
/// [`begin`](Self::begin) for explicit pending/authorized states, then install
/// the same value as both [`TokenProvider`] and [`OAuthScopeEscalationHandler`].
#[derive(Clone)]
pub struct OAuthAuthorizationFlow {
    inner: Arc<FlowInner>,
}

impl fmt::Debug for OAuthAuthorizationFlow {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OAuthAuthorizationFlow")
            .field("resource_url", &self.inner.resource_url)
            .field("resource_override", &self.inner.resource_override)
            .field("redirect_policy", &self.inner.redirect_policy)
            .field("preferred_issuer", &self.inner.preferred_issuer)
            .finish_non_exhaustive()
    }
}

/// Builder for [`OAuthAuthorizationFlow`].
pub struct OAuthAuthorizationFlowBuilder {
    resource_url: String,
    resource_override: Option<String>,
    registration_options: OAuthClientRegistrationOptions,
    pre_registered_client: Option<(String, Option<String>)>,
    registration_store: Option<Arc<dyn OAuthClientRegistrationStore>>,
    token_store: Option<Arc<dyn OAuthTokenStore>>,
    state_store: Option<Arc<dyn OAuthAuthorizationStateStore>>,
    http: Option<Arc<dyn OAuthHttpClient>>,
    redirect_policy: Option<OAuthRedirectPolicy>,
    preferred_issuer: Option<String>,
    refresh_buffer: Duration,
    authorization_handler: Option<Arc<dyn OAuthAuthorizationHandler>>,
    assertion_signer: Option<Arc<dyn OAuthClientAssertionSigner>>,
}

impl fmt::Debug for OAuthAuthorizationFlowBuilder {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OAuthAuthorizationFlowBuilder")
            .field("resource_url", &self.resource_url)
            .field("resource_override", &self.resource_override)
            .field("registration_options", &self.registration_options)
            .field("redirect_policy", &self.redirect_policy)
            .field("preferred_issuer", &self.preferred_issuer)
            .field("refresh_buffer", &self.refresh_buffer)
            .finish_non_exhaustive()
    }
}

impl OAuthAuthorizationFlowBuilder {
    /// Configure available client-registration mechanisms.
    pub fn registration_options(mut self, options: OAuthClientRegistrationOptions) -> Self {
        self.registration_options = options;
        self
    }

    /// Configure pre-registered credentials and bind them to the issuer selected
    /// during discovery.
    ///
    /// This is the convenient form for applications that know their client
    /// credentials before they know which authorization-server issuer the
    /// protected resource will advertise. It takes priority over CIMD and DCR,
    /// just like [`OAuthClientRegistrationOptions::with_pre_registered`].
    pub fn pre_registered_client(
        mut self,
        client_id: impl Into<String>,
        client_secret: Option<String>,
    ) -> Self {
        self.pre_registered_client = Some((client_id.into(), client_secret));
        self
    }

    /// Configure issuer-bound client-credential persistence.
    pub fn registration_store(
        mut self,
        store: impl OAuthClientRegistrationStore + 'static,
    ) -> Self {
        self.registration_store = Some(Arc::new(store));
        self
    }

    /// Configure token persistence.
    pub fn token_store(mut self, store: impl OAuthTokenStore + 'static) -> Self {
        self.token_store = Some(Arc::new(store));
        self
    }

    /// Configure persisted PKCE and CSRF state.
    pub fn state_store(mut self, store: impl OAuthAuthorizationStateStore + 'static) -> Self {
        self.state_store = Some(Arc::new(store));
        self
    }

    /// Configure the OAuth protocol HTTP implementation.
    pub fn http_client(mut self, client: impl OAuthHttpClient) -> Self {
        self.http = Some(Arc::new(client));
        self
    }

    /// Configure explicit redirect handling.
    pub fn redirect_policy(mut self, policy: OAuthRedirectPolicy) -> Self {
        self.redirect_policy = Some(policy);
        self
    }

    /// Select one exact issuer when the resource advertises multiple servers.
    pub fn preferred_authorization_server(mut self, issuer: impl Into<String>) -> Self {
        self.preferred_issuer = Some(issuer.into());
        self
    }

    /// Override the RFC 8707 resource indicator.
    ///
    /// When set, this value replaces the resource discovered from Protected
    /// Resource Metadata everywhere the resource indicator is sent: the
    /// authorize URL, code exchange, and refresh requests. It is also what
    /// the [`OAuthTokenBinding`] is keyed on. Discovery still runs to locate
    /// the authorization server; only the resource indicator used in token
    /// requests is overridden. Must be an absolute URL without a fragment,
    /// the same requirement enforced on the discovered resource.
    pub fn resource(mut self, resource: impl Into<String>) -> Self {
        self.resource_override = Some(resource.into());
        self
    }

    /// Set the pre-expiry refresh buffer.
    pub fn refresh_buffer(mut self, buffer: Duration) -> Self {
        self.refresh_buffer = buffer;
        self
    }

    /// Configure automatic browser/headless authorization handling.
    pub fn authorization_handler(mut self, handler: impl OAuthAuthorizationHandler) -> Self {
        self.authorization_handler = Some(Arc::new(handler));
        self
    }

    /// Configure `private_key_jwt` assertion signing.
    pub fn client_assertion_signer(mut self, signer: impl OAuthClientAssertionSigner) -> Self {
        self.assertion_signer = Some(Arc::new(signer));
        self
    }

    /// Build the state machine.
    ///
    /// # Errors
    ///
    /// Returns an error when no redirect policy was supplied or when the
    /// default reqwest client cannot be constructed.
    pub fn build(self) -> Result<OAuthAuthorizationFlow, OAuthClientError> {
        let redirect_policy = self.redirect_policy.ok_or_else(|| {
            OAuthClientError::BuildError(
                "OAuthAuthorizationFlow requires an explicit redirect policy".to_string(),
            )
        })?;
        validate_redirect_policy(&redirect_policy)?;
        if let Some(resource) = &self.resource_override {
            validate_resource_override(resource)?;
        }
        let http = match self.http {
            Some(http) => http,
            None => Arc::new(ReqwestOAuthHttpClient::without_redirects()?),
        };
        Ok(OAuthAuthorizationFlow {
            inner: Arc::new(FlowInner {
                resource_url: self.resource_url,
                resource_override: self.resource_override,
                registration_options: self.registration_options,
                pre_registered_client: self.pre_registered_client,
                registration_store: self.registration_store.unwrap_or_else(|| {
                    Arc::new(super::oauth_authcode::MemoryOAuthClientRegistrationStore::new())
                }),
                token_store: self
                    .token_store
                    .unwrap_or_else(|| Arc::new(MemoryOAuthTokenStore::new())),
                state_store: self
                    .state_store
                    .unwrap_or_else(|| Arc::new(MemoryOAuthAuthorizationStateStore::new())),
                http,
                redirect_policy,
                preferred_issuer: self.preferred_issuer,
                refresh_buffer: self.refresh_buffer,
                authorization_handler: self.authorization_handler,
                assertion_signer: self.assertion_signer,
                current: RwLock::new(None),
                authorization_lock: Mutex::new(()),
                refresh_lock: Mutex::new(()),
            }),
        })
    }
}

fn validate_redirect_policy(policy: &OAuthRedirectPolicy) -> Result<(), OAuthClientError> {
    let (redirect_uri, callback_path) = match policy {
        OAuthRedirectPolicy::Fixed { redirect_uri } => (Some(redirect_uri.as_str()), None),
        OAuthRedirectPolicy::Loopback { callback_path, .. } => (None, Some(callback_path)),
    };
    if let Some(uri) = redirect_uri {
        let parsed = reqwest::Url::parse(uri)
            .map_err(|error| OAuthClientError::BuildError(error.to_string()))?;
        if parsed.fragment().is_some() {
            return Err(OAuthClientError::BuildError(
                "OAuth redirect URI must not contain a fragment".to_string(),
            ));
        }
    }
    if let Some(path) = callback_path
        && (!path.starts_with('/') || path.contains('?') || path.contains('#'))
    {
        return Err(OAuthClientError::BuildError(
            "loopback callback path must begin with `/` and contain no query or fragment"
                .to_string(),
        ));
    }
    Ok(())
}

/// Result of beginning an OAuth authorization attempt.
#[non_exhaustive]
pub enum OAuthAuthorizationStart {
    /// A persisted, sufficiently scoped token was restored without redirecting.
    Authorized {
        /// Scopes restored from the token store.
        scopes: Vec<String>,
    },
    /// User-agent authorization is required.
    Pending(OAuthPendingAuthorization),
}

impl fmt::Debug for OAuthAuthorizationStart {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Authorized { scopes } => formatter
                .debug_struct("Authorized")
                .field("scopes", scopes)
                .finish(),
            Self::Pending(pending) => formatter.debug_tuple("Pending").field(pending).finish(),
        }
    }
}

/// An authorization attempt waiting for a callback.
pub struct OAuthPendingAuthorization {
    flow: OAuthAuthorizationFlow,
    request: OAuthAuthorizationRequest,
    state: String,
    callback_rx: Mutex<Option<oneshot::Receiver<String>>>,
}

impl fmt::Debug for OAuthPendingAuthorization {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OAuthPendingAuthorization")
            .field("request", &self.request)
            .field("state", &"[REDACTED]")
            .finish_non_exhaustive()
    }
}

impl OAuthPendingAuthorization {
    /// Authorization request to present to a user agent or headless harness.
    pub fn request(&self) -> &OAuthAuthorizationRequest {
        &self.request
    }

    /// Complete this attempt with the callback URL returned by the AS.
    pub async fn complete_callback_url(self, callback_url: &str) -> Result<(), OAuthClientError> {
        self.flow
            .complete_callback_url_for_state(callback_url, Some(&self.state))
            .await
    }

    /// Wait for a configured loopback redirect and complete the attempt.
    pub async fn wait_for_callback(self) -> Result<(), OAuthClientError> {
        self.wait_for_callback_with_timeout(Duration::from_secs(300))
            .await
    }

    /// Wait for a loopback redirect with a custom timeout.
    pub async fn wait_for_callback_with_timeout(
        self,
        timeout: Duration,
    ) -> Result<(), OAuthClientError> {
        let receiver = self.callback_rx.lock().await.take().ok_or_else(|| {
            OAuthClientError::Redirect(
                "this redirect policy does not own a loopback callback".to_string(),
            )
        })?;
        let callback_url = tokio::time::timeout(timeout, receiver)
            .await
            .map_err(|_| OAuthClientError::Redirect("OAuth callback timed out".to_string()))?
            .map_err(|_| {
                OAuthClientError::Redirect("OAuth callback listener closed".to_string())
            })?;
        self.flow
            .complete_callback_url_for_state(&callback_url, Some(&self.state))
            .await
    }
}

impl OAuthAuthorizationFlow {
    /// Begin building a state machine for `resource_url`.
    pub fn builder(resource_url: impl Into<String>) -> OAuthAuthorizationFlowBuilder {
        OAuthAuthorizationFlowBuilder {
            resource_url: resource_url.into(),
            resource_override: None,
            registration_options: OAuthClientRegistrationOptions::new(),
            pre_registered_client: None,
            registration_store: None,
            token_store: None,
            state_store: None,
            http: None,
            redirect_policy: None,
            preferred_issuer: None,
            refresh_buffer: Duration::from_secs(30),
            authorization_handler: None,
            assertion_signer: None,
        }
    }

    /// Discover, register, restore or create authorization state, and return
    /// the next explicit state.
    pub async fn begin<I, S>(&self, scopes: I) -> Result<OAuthAuthorizationStart, OAuthClientError>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        self.begin_with_challenge(unique_scopes(scopes), None).await
    }

    /// Drive a complete authorization using the configured
    /// [`OAuthAuthorizationHandler`].
    pub async fn authorize<I, S>(&self, scopes: I) -> Result<(), OAuthClientError>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        self.authorize_with_challenge(unique_scopes(scopes), None)
            .await
    }

    async fn authorize_with_challenge(
        &self,
        scopes: Vec<String>,
        challenge: Option<OAuthBearerChallenge>,
    ) -> Result<(), OAuthClientError> {
        let _guard = self.inner.authorization_lock.lock().await;
        let start = self.begin_with_challenge(scopes, challenge).await?;
        let OAuthAuthorizationStart::Pending(pending) = start else {
            return Ok(());
        };
        let handler = self.inner.authorization_handler.as_ref().ok_or_else(|| {
            OAuthClientError::Redirect(
                "authorization is pending; configure an OAuthAuthorizationHandler or use begin()"
                    .to_string(),
            )
        })?;
        match handler.authorize(pending.request().clone()).await? {
            OAuthAuthorizationAction::AwaitLoopback => pending.wait_for_callback().await,
            OAuthAuthorizationAction::CallbackUrl(url) => pending.complete_callback_url(&url).await,
        }
    }

    async fn begin_with_challenge(
        &self,
        explicit_scopes: Vec<String>,
        challenge: Option<OAuthBearerChallenge>,
    ) -> Result<OAuthAuthorizationStart, OAuthClientError> {
        let discovery = discover_with_http(
            &self.inner.resource_url,
            challenge,
            self.inner.http.as_ref(),
        )
        .await?;
        let metadata =
            select_authorization_server(&discovery, self.inner.preferred_issuer.as_deref())?;
        require_s256(&metadata)?;

        let state = random_urlsafe(16);
        let code_verifier = random_urlsafe(32);
        let code_challenge = pkce_challenge(&code_verifier);
        let (redirect_uri, callback_rx) =
            prepare_redirect(&self.inner.redirect_policy, &state).await?;

        let mut registration_options = self.inner.registration_options.clone();
        if registration_options.pre_registered.is_none()
            && let Some((client_id, client_secret)) = &self.inner.pre_registered_client
        {
            registration_options.pre_registered = Some(OAuthClientRegistration::pre_registered(
                metadata.issuer.clone(),
                client_id.clone(),
                client_secret.clone(),
            ));
        }
        if let Some(dynamic) = registration_options.dynamic_registration.as_mut() {
            dynamic.redirect_uris = vec![redirect_uri.clone()];
            if metadata
                .grant_types_supported
                .iter()
                .any(|grant| grant == "refresh_token")
                && !dynamic
                    .grant_types
                    .iter()
                    .any(|grant| grant == "refresh_token")
            {
                dynamic.grant_types.push("refresh_token".to_string());
            }
            dynamic.token_endpoint_auth_method = preferred_registration_auth_method(
                &metadata.token_endpoint_auth_methods_supported,
                self.inner.assertion_signer.is_some(),
            )
            .to_string();
        }
        let registration = resolve_registration_with_http(
            self.inner.http.as_ref(),
            &metadata,
            &registration_options,
            self.inner.registration_store.as_ref(),
        )
        .await?;

        let auth_method = OAuthTokenEndpointAuthMethod::select_with_private_key(
            &metadata.token_endpoint_auth_methods_supported,
            registration.client_secret().is_some(),
            self.inner.assertion_signer.is_some(),
        )?;
        let scopes = select_scopes(&explicit_scopes, &discovery, &metadata);
        // The explicit override, when configured, replaces the discovered
        // resource everywhere the RFC 8707 resource indicator is sent, and
        // is what the token binding below is keyed on.
        let resource = self
            .inner
            .resource_override
            .clone()
            .unwrap_or_else(|| discovery.resource.clone());
        let binding = OAuthTokenBinding {
            resource: resource.clone(),
            issuer: metadata.issuer.clone(),
            client_id: registration.client_id().to_string(),
        };

        if let Some(token) = self.inner.token_store.load(&binding).await?
            && scopes_are_covered(&scopes, &token.scopes)
            && (token_is_valid(&token, self.inner.refresh_buffer) || token.refresh_token.is_some())
        {
            *self.inner.current.write().await = Some(ActiveToken {
                binding,
                token: token.clone(),
                token_endpoint: metadata.token_endpoint.clone(),
                registration: registration.clone(),
                auth_method,
            });
            if token_is_valid(&token, self.inner.refresh_buffer) {
                return Ok(OAuthAuthorizationStart::Authorized {
                    scopes: token.scopes,
                });
            }
            match TokenProvider::get_token(self).await {
                Ok(_) => {
                    return Ok(OAuthAuthorizationStart::Authorized {
                        scopes: token.scopes,
                    });
                }
                Err(OAuthClientError::TokenRefreshRejected(_)) => {
                    // get_token() already cleared `current` and removed the
                    // stored token; treat the stored token as unusable and
                    // fall through to a new authorization request instead of
                    // failing begin() on every subsequent call.
                }
                Err(error) => return Err(error),
            }
        }

        let pending_state = OAuthPendingAuthorizationState {
            state: state.clone(),
            code_verifier,
            redirect_uri: redirect_uri.clone(),
            resource: resource.clone(),
            issuer: metadata.issuer.clone(),
            token_endpoint: metadata.token_endpoint.clone(),
            registration: registration.clone(),
            token_endpoint_auth_method: auth_method,
            scopes: scopes.clone(),
            iss_required: metadata.authorization_response_iss_parameter_supported,
        };
        self.inner.state_store.save(&pending_state).await?;

        let mut authorization_url = reqwest::Url::parse(&metadata.authorization_endpoint)
            .map_err(|error| OAuthClientError::Discovery(error.to_string()))?;
        {
            let mut query = authorization_url.query_pairs_mut();
            query
                .append_pair("response_type", "code")
                .append_pair("client_id", registration.client_id())
                .append_pair("redirect_uri", &redirect_uri)
                .append_pair("state", &state)
                .append_pair("code_challenge", &code_challenge)
                .append_pair("code_challenge_method", "S256")
                .append_pair("resource", &resource);
            if !scopes.is_empty() {
                query.append_pair("scope", &scopes.join(" "));
            }
        }
        Ok(OAuthAuthorizationStart::Pending(
            OAuthPendingAuthorization {
                flow: self.clone(),
                request: OAuthAuthorizationRequest {
                    authorization_url: authorization_url.to_string(),
                    redirect_uri,
                    resource,
                    issuer: metadata.issuer,
                    scopes,
                },
                state,
                callback_rx: Mutex::new(callback_rx),
            },
        ))
    }

    /// Resume a persisted authorization attempt from a callback URL.
    pub async fn complete_callback_url(&self, callback_url: &str) -> Result<(), OAuthClientError> {
        self.complete_callback_url_for_state(callback_url, None)
            .await
    }

    async fn complete_callback_url_for_state(
        &self,
        callback_url: &str,
        expected_state: Option<&str>,
    ) -> Result<(), OAuthClientError> {
        let parsed = parse_callback_url(callback_url)?;
        if let Some(expected) = expected_state
            && parsed.state != expected
        {
            return Err(OAuthClientError::InvalidResponse(
                "OAuth callback state mismatch".to_string(),
            ));
        }
        let pending = self
            .inner
            .state_store
            .load(&parsed.state)
            .await?
            .ok_or_else(|| {
                OAuthClientError::StateStore(
                    "no persisted OAuth authorization matches the callback state".to_string(),
                )
            })?;
        validate_callback_target(callback_url, &pending.redirect_uri)?;
        validate_callback_issuer(
            parsed.issuer.as_deref(),
            &pending.issuer,
            pending.iss_required,
        )?;

        let fields = vec![
            ("grant_type".to_string(), "authorization_code".to_string()),
            ("code".to_string(), parsed.code),
            ("redirect_uri".to_string(), pending.redirect_uri.clone()),
            ("code_verifier".to_string(), pending.code_verifier.clone()),
            ("resource".to_string(), pending.resource.clone()),
        ];
        let response = send_token_request(
            self.inner.http.as_ref(),
            &pending.token_endpoint,
            &pending.issuer,
            &pending.registration,
            pending.token_endpoint_auth_method,
            fields,
            self.inner.assertion_signer.as_deref(),
        )
        .await?;
        let token = token_from_response(response, &pending.scopes, None)?;
        let binding = OAuthTokenBinding {
            resource: pending.resource.clone(),
            issuer: pending.issuer.clone(),
            client_id: pending.registration.client_id().to_string(),
        };
        self.inner.token_store.save(&binding, &token).await?;
        self.inner.state_store.remove(&parsed.state).await?;
        *self.inner.current.write().await = Some(ActiveToken {
            binding,
            token,
            token_endpoint: pending.token_endpoint,
            registration: pending.registration,
            auth_method: pending.token_endpoint_auth_method,
        });
        Ok(())
    }

    /// Return the currently authorized scope set, if any.
    pub async fn authorized_scopes(&self) -> Option<Vec<String>> {
        self.inner
            .current
            .read()
            .await
            .as_ref()
            .map(|active| active.token.scopes.clone())
    }
}

#[async_trait]
impl TokenProvider for OAuthAuthorizationFlow {
    async fn get_token(&self) -> Result<String, OAuthClientError> {
        if let Some(active) = self.inner.current.read().await.as_ref()
            && token_is_valid(&active.token, self.inner.refresh_buffer)
        {
            return Ok(active.token.access_token.clone());
        }

        let _guard = self.inner.refresh_lock.lock().await;
        let active = self.inner.current.read().await.clone().ok_or_else(|| {
            OAuthClientError::TokenRequest(
                "OAuthAuthorizationFlow is not authorized; call authorize() or begin()".to_string(),
            )
        })?;
        if token_is_valid(&active.token, self.inner.refresh_buffer) {
            return Ok(active.token.access_token);
        }

        // Another flow instance, in this process or another, may share this
        // binding's token store. If it already refreshed, our in-memory
        // copy is stale; reload the store before spending a network round
        // trip. Adopting a different, still-valid stored token here also
        // avoids handing an authorization server that rotates refresh
        // tokens on use a refresh token it has already consumed, which it
        // would answer with invalid_grant (#1472).
        if let Some(stored) = self.inner.token_store.load(&active.binding).await?
            && tokens_differ(&stored, &active.token)
            && token_is_valid(&stored, self.inner.refresh_buffer)
        {
            let access_token = stored.access_token.clone();
            *self.inner.current.write().await = Some(ActiveToken {
                token: stored,
                ..active
            });
            return Ok(access_token);
        }

        let refresh_token = active.token.refresh_token.as_deref().ok_or_else(|| {
            OAuthClientError::TokenRequest(
                "OAuth access token expired and no refresh token is available; call authorize() \
                 or begin() again to reauthorize"
                    .to_string(),
            )
        })?;
        let mut fields = vec![
            ("grant_type".to_string(), "refresh_token".to_string()),
            ("refresh_token".to_string(), refresh_token.to_string()),
            ("resource".to_string(), active.binding.resource.clone()),
        ];
        if !active.token.scopes.is_empty() {
            fields.push(("scope".to_string(), active.token.scopes.join(" ")));
        }
        let response = send_token_request(
            self.inner.http.as_ref(),
            &active.token_endpoint,
            &active.binding.issuer,
            &active.registration,
            active.auth_method,
            fields,
            self.inner.assertion_signer.as_deref(),
        )
        .await;
        let response = match response {
            Ok(response) => response,
            Err(error @ OAuthClientError::TokenRefreshRejected(_)) => {
                // The refresh token we sent may already have been consumed
                // by another flow sharing this store: an authorization
                // server that rotates refresh tokens on use accepts only
                // the first use and answers invalid_grant to the rest.
                // Reload once before giving up; if the store now holds a
                // different, valid token for this binding, adopt it instead
                // of discarding and forcing reauthorization (#1472).
                let restored = self.inner.token_store.load(&active.binding).await?;
                if let Some(stored) = &restored
                    && tokens_differ(stored, &active.token)
                    && token_is_valid(stored, self.inner.refresh_buffer)
                {
                    let access_token = stored.access_token.clone();
                    *self.inner.current.write().await = Some(ActiveToken {
                        token: stored.clone(),
                        ..active
                    });
                    return Ok(access_token);
                }
                // The authorization server revoked or expired the refresh
                // token, and the store does not hold a newer one either.
                // Discard it so the next begin() reauthorizes instead of
                // retrying a dead token on every call. Other failures
                // (network errors, 5xx) are left in place; they may be
                // transient.
                *self.inner.current.write().await = None;
                // Only remove the store entry if it still matches the token
                // we just tried; a concurrent flow may already have saved a
                // fresh one there, and removing it here would discard a
                // token that is still good out from under it.
                let discard = match &restored {
                    Some(stored) => !tokens_differ(stored, &active.token),
                    None => true,
                };
                if discard {
                    self.inner.token_store.remove(&active.binding).await?;
                }
                return Err(error);
            }
            Err(error) => return Err(error),
        };
        let token = token_from_response(
            response,
            &active.token.scopes,
            active.token.refresh_token.clone(),
        )?;
        self.inner.token_store.save(&active.binding, &token).await?;
        let access_token = token.access_token.clone();
        *self.inner.current.write().await = Some(ActiveToken { token, ..active });
        Ok(access_token)
    }

    /// Mark the current access token unusable while keeping the refresh
    /// token, so the next [`get_token`](Self::get_token) call refreshes
    /// instead of re-sending a token the server just rejected.
    ///
    /// This only touches the in-memory `current` token; nothing is written
    /// to the token store here. `get_token`'s refresh path already persists
    /// the replacement the same way an ordinary pre-expiry refresh does, so
    /// there is nothing left to do once that refresh lands. With no refresh
    /// token, the next `get_token` call reports that reauthorization is
    /// needed instead of retrying a token that cannot be renewed.
    ///
    /// A concurrent refresh can race this: `get_token` may already be
    /// mid-refresh under `refresh_lock` and overwrite `current` with a fresh
    /// token right after this write. That costs one wasted refresh at worst,
    /// not correctness, so this does not take `refresh_lock` itself (nor
    /// hold `current`'s write lock across an await).
    async fn invalidate(&self) {
        if let Some(active) = self.inner.current.write().await.as_mut() {
            active.token.expires_at = 0;
        }
    }
}

#[async_trait]
impl OAuthScopeEscalationHandler for OAuthAuthorizationFlow {
    async fn reauthorize(
        &self,
        request: OAuthScopeEscalationRequest,
    ) -> Result<(), OAuthClientError> {
        if request.resource != self.inner.resource_url {
            return Err(OAuthClientError::ScopeEscalation(format!(
                "scope challenge resource `{}` does not match flow resource `{}`",
                request.resource, self.inner.resource_url
            )));
        }
        let challenge = OAuthBearerChallenge {
            error: Some("insufficient_scope".to_string()),
            scopes: request.challenge.required_scopes,
            resource_metadata: request.challenge.resource_metadata,
            error_description: request.challenge.error_description,
        };
        self.authorize_with_challenge(request.requested_scopes, Some(challenge))
            .await
            .map_err(|error| OAuthClientError::ScopeEscalation(error.to_string()))
    }
}

#[derive(Debug)]
struct FlowDiscovery {
    resource: String,
    protected_resource: OAuthProtectedResourceMetadata,
    authorization_servers: Vec<OAuthAuthorizationServerMetadata>,
    challenge: Option<OAuthBearerChallenge>,
}

async fn discover_with_http(
    resource_url: &str,
    challenge: Option<OAuthBearerChallenge>,
    http: &dyn OAuthHttpClient,
) -> Result<FlowDiscovery, OAuthClientError> {
    let challenge = match challenge {
        Some(challenge) => Some(challenge),
        None => {
            // Probe a protected MCP operation rather than `initialize`, which
            // servers commonly leave public. A POST also works with MCP
            // endpoints that do not implement GET and therefore expose their
            // RFC 9728 discovery hint only on normal protocol requests.
            let mut request = OAuthHttpRequest::post_json(
                resource_url,
                serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": "tower-mcp-oauth-discovery",
                    "method": "tools/list",
                    "params": {}
                }),
            );
            request.headers.push((
                "accept".to_string(),
                "application/json, text/event-stream".to_string(),
            ));
            let response = http.execute(request).await?;
            response
                .header_values("www-authenticate")
                .find_map(OAuthBearerChallenge::from_www_authenticate)
        }
    };
    let protected_resource =
        discover_protected_resource_metadata(resource_url, &challenge, http).await?;
    if protected_resource.authorization_servers.is_empty() {
        return Err(OAuthClientError::Discovery(
            "protected resource metadata omitted authorization_servers".to_string(),
        ));
    }

    let mut authorization_servers = Vec::new();
    let mut last_error = None;
    for issuer in &protected_resource.authorization_servers {
        match discover_authorization_server(http, issuer).await {
            Ok(metadata) => authorization_servers.push(metadata),
            Err(error) => last_error = Some(error),
        }
    }
    if authorization_servers.is_empty() {
        return Err(last_error.unwrap_or_else(|| {
            OAuthClientError::Discovery(
                "protected resource advertised no usable authorization server".to_string(),
            )
        }));
    }
    Ok(FlowDiscovery {
        resource: protected_resource.resource.clone(),
        protected_resource,
        authorization_servers,
        challenge,
    })
}

/// Discover Protected Resource Metadata for `resource_url`, trying
/// candidates in order and skipping any that fail to fetch, fail to parse,
/// or whose `resource` does not cover `resource_url` (see
/// [`validate_resource_identifier`]):
///
/// 1. The challenge's `resource_metadata` URL, when present. Rejecting this
///    candidate does not fail discovery; it falls through to the well-known
///    candidates below instead.
/// 2. The final path-aware RFC 9728 well-known location.
/// 3. The origin root well-known location, for compatibility.
///
/// If every candidate is rejected, the returned error names each candidate
/// URL tried and why it was rejected.
async fn discover_protected_resource_metadata(
    resource_url: &str,
    challenge: &Option<OAuthBearerChallenge>,
    http: &dyn OAuthHttpClient,
) -> Result<OAuthProtectedResourceMetadata, OAuthClientError> {
    let mut attempts = Vec::new();

    if let Some(url) = challenge
        .as_ref()
        .and_then(|challenge| challenge.resource_metadata.as_deref())
    {
        match fetch_matching_protected_resource(http, url, resource_url).await {
            Ok(metadata) => return Ok(metadata),
            Err(reason) => {
                tracing::debug!(
                    %url,
                    %reason,
                    "OAuth protected-resource metadata from challenge failed; falling back to well-known candidates"
                );
                attempts.push(format!("`{url}` (from challenge): {reason}"));
            }
        }
    }

    for url in protected_resource_metadata_urls(resource_url)? {
        match fetch_matching_protected_resource(http, &url, resource_url).await {
            Ok(metadata) => return Ok(metadata),
            Err(reason) => {
                tracing::debug!(%url, %reason, "OAuth protected-resource metadata candidate failed");
                attempts.push(format!("`{url}`: {reason}"));
            }
        }
    }

    Err(OAuthClientError::Discovery(format!(
        "could not discover Protected Resource Metadata for `{resource_url}`; tried {}",
        attempts.join(", ")
    )))
}

/// Fetch one Protected Resource Metadata candidate and confirm its
/// `resource` covers `resource_url` per [`validate_resource_identifier`].
/// Returns a plain-text rejection reason rather than [`OAuthClientError`] so
/// [`discover_protected_resource_metadata`] can fold several attempts into
/// one aggregate discovery error.
async fn fetch_matching_protected_resource(
    http: &dyn OAuthHttpClient,
    url: &str,
    resource_url: &str,
) -> Result<OAuthProtectedResourceMetadata, String> {
    let metadata = fetch_json::<OAuthProtectedResourceMetadata>(http, url)
        .await
        .map_err(discovery_reason)?;
    validate_resource_identifier(resource_url, &metadata.resource).map_err(discovery_reason)?;
    Ok(metadata)
}

async fn discover_authorization_server(
    http: &dyn OAuthHttpClient,
    issuer: &str,
) -> Result<OAuthAuthorizationServerMetadata, OAuthClientError> {
    let mut last_error = None;
    for url in authorization_server_metadata_urls(issuer)? {
        match fetch_json::<OAuthAuthorizationServerMetadata>(http, &url).await {
            Ok(metadata) if metadata.issuer == issuer => return Ok(metadata),
            Ok(metadata) => {
                last_error = Some(OAuthClientError::Discovery(format!(
                    "authorization server issuer mismatch: expected `{issuer}`, got `{}`",
                    metadata.issuer
                )))
            }
            Err(error) => last_error = Some(error),
        }
    }
    Err(last_error.unwrap_or_else(|| {
        OAuthClientError::Discovery(format!(
            "could not discover authorization server metadata for `{issuer}`"
        ))
    }))
}

async fn fetch_json<T: serde::de::DeserializeOwned>(
    http: &dyn OAuthHttpClient,
    url: &str,
) -> Result<T, OAuthClientError> {
    let response = http.execute(OAuthHttpRequest::get(url)).await?;
    if !response.is_success() {
        return Err(OAuthClientError::Discovery(format!(
            "GET `{url}` returned HTTP {}: {}",
            response.status,
            response.error_body()
        )));
    }
    response.json()
}

fn protected_resource_metadata_urls(resource_url: &str) -> Result<Vec<String>, OAuthClientError> {
    let parsed = reqwest::Url::parse(resource_url)
        .map_err(|error| OAuthClientError::Discovery(error.to_string()))?;
    let origin = parsed.origin().ascii_serialization();
    let path = parsed.path();
    let mut urls = Vec::new();
    if !path.is_empty() && path != "/" {
        urls.push(format!(
            "{origin}/.well-known/oauth-protected-resource{path}"
        ));
    }
    urls.push(format!("{origin}/.well-known/oauth-protected-resource"));
    urls.dedup();
    Ok(urls)
}

fn authorization_server_metadata_urls(issuer: &str) -> Result<Vec<String>, OAuthClientError> {
    let parsed = reqwest::Url::parse(issuer)
        .map_err(|error| OAuthClientError::Discovery(error.to_string()))?;
    let origin = parsed.origin().ascii_serialization();
    let path = parsed.path();
    let trimmed = issuer.trim_end_matches('/');
    let mut urls = Vec::new();
    if !path.is_empty() && path != "/" {
        urls.push(format!(
            "{origin}/.well-known/oauth-authorization-server{path}"
        ));
        urls.push(format!("{origin}/.well-known/openid-configuration{path}"));
        urls.push(format!("{trimmed}/.well-known/openid-configuration"));
    } else {
        urls.push(format!("{origin}/.well-known/oauth-authorization-server"));
        urls.push(format!("{origin}/.well-known/openid-configuration"));
    }
    urls.dedup();
    Ok(urls)
}

fn select_authorization_server(
    discovery: &FlowDiscovery,
    preferred_issuer: Option<&str>,
) -> Result<OAuthAuthorizationServerMetadata, OAuthClientError> {
    match preferred_issuer {
        Some(issuer) => discovery
            .authorization_servers
            .iter()
            .find(|metadata| metadata.issuer == issuer)
            .cloned()
            .ok_or_else(|| {
                OAuthClientError::Discovery(format!(
                    "preferred authorization server `{issuer}` was not advertised"
                ))
            }),
        None => discovery
            .authorization_servers
            .first()
            .cloned()
            .ok_or_else(|| OAuthClientError::Discovery("no authorization server found".into())),
    }
}

fn require_s256(metadata: &OAuthAuthorizationServerMetadata) -> Result<(), OAuthClientError> {
    if metadata
        .code_challenge_methods_supported
        .iter()
        .any(|method| method == "S256")
    {
        Ok(())
    } else {
        Err(OAuthClientError::Discovery(format!(
            "authorization server `{}` does not advertise PKCE S256",
            metadata.issuer
        )))
    }
}

fn select_scopes(
    explicit: &[String],
    discovery: &FlowDiscovery,
    metadata: &OAuthAuthorizationServerMetadata,
) -> Vec<String> {
    let selected = if !explicit.is_empty() {
        explicit.to_vec()
    } else if let Some(challenge) = &discovery.challenge
        && !challenge.scopes.is_empty()
    {
        challenge.scopes.clone()
    } else {
        // MCP scope selection strategy: challenge scope, else the resource's
        // `scopes_supported`, else omit `scope`. The authorization server's own
        // `scopes_supported` describes everything it issues, not what this
        // resource needs, so it is not a fallback.
        discovery.protected_resource.scopes_supported.clone()
    };
    let mut selected = unique_scopes(selected);
    let refresh_supported = metadata.grant_types_supported.is_empty()
        || metadata
            .grant_types_supported
            .iter()
            .any(|grant| grant == "refresh_token");
    if refresh_supported
        && metadata
            .scopes_supported
            .iter()
            .any(|scope| scope == "offline_access")
        && !selected.iter().any(|scope| scope == "offline_access")
    {
        selected.push("offline_access".to_string());
    }
    selected
}

async fn resolve_registration_with_http(
    http: &dyn OAuthHttpClient,
    metadata: &OAuthAuthorizationServerMetadata,
    options: &OAuthClientRegistrationOptions,
    store: &dyn OAuthClientRegistrationStore,
) -> Result<OAuthClientRegistration, OAuthClientError> {
    if let Some(registration) = &options.pre_registered {
        if registration.method() != OAuthClientRegistrationMethod::PreRegistered {
            return Err(OAuthClientError::BuildError(
                "pre_registered must contain pre-registered credentials".to_string(),
            ));
        }
        if registration.bound_issuer() != Some(metadata.issuer.as_str()) {
            return Err(OAuthClientError::BuildError(format!(
                "pre-registered credentials are bound to issuer {:?}, not `{}`",
                registration.bound_issuer(),
                metadata.issuer
            )));
        }
        return Ok(registration.clone());
    }

    if metadata.client_id_metadata_document_supported
        && let Some(client_id) = &options.client_id_metadata_document
    {
        validate_cimd_url(client_id)?;
        return Ok(OAuthClientRegistration::client_id_metadata_document(
            client_id.clone(),
        ));
    }

    if options.dynamic_registration.is_some()
        && let Some(registration) = store.load(&metadata.issuer).await?
    {
        if registration.method() != OAuthClientRegistrationMethod::Dynamic
            || registration.bound_issuer() != Some(metadata.issuer.as_str())
        {
            return Err(OAuthClientError::CredentialStore(format!(
                "stored registration is not dynamically bound to `{}`",
                metadata.issuer
            )));
        }
        return Ok(registration);
    }

    if let (Some(endpoint), Some(request)) = (
        metadata.registration_endpoint.as_deref(),
        options.dynamic_registration.as_ref(),
    ) {
        if request.redirect_uris.is_empty() {
            return Err(OAuthClientError::BuildError(
                "dynamic registration requires a redirect URI".to_string(),
            ));
        }
        let value = serde_json::to_value(request)
            .map_err(|error| OAuthClientError::Registration(error.to_string()))?;
        let response = http
            .execute(OAuthHttpRequest::post_json(endpoint, value))
            .await?;
        if !response.is_success() {
            return Err(OAuthClientError::Registration(format!(
                "dynamic registration returned HTTP {}: {}",
                response.status,
                response.error_body()
            )));
        }
        #[derive(serde::Deserialize)]
        struct RegistrationResponse {
            client_id: String,
            client_secret: Option<String>,
            token_endpoint_auth_method: Option<String>,
        }
        let registered: RegistrationResponse = response.json()?;
        // RFC 7591 section 3.2.1: the response reports the registered
        // method, which can differ from the requested one.
        require_secret_for_confidential_registration(
            registered
                .token_endpoint_auth_method
                .as_deref()
                .unwrap_or(&request.token_endpoint_auth_method),
            registered.client_secret.as_deref(),
        )?;
        let registration = OAuthClientRegistration::dynamically_registered(
            metadata.issuer.clone(),
            registered.client_id,
            registered.client_secret,
        );
        store.save(&metadata.issuer, &registration).await?;
        return Ok(registration);
    }

    Err(OAuthClientError::BuildError(
        "authorization server supports none of the configured client registration mechanisms"
            .to_string(),
    ))
}

fn validate_cimd_url(client_id: &str) -> Result<(), OAuthClientError> {
    let url = reqwest::Url::parse(client_id).map_err(|error| {
        OAuthClientError::BuildError(format!("invalid CIMD client ID `{client_id}`: {error}"))
    })?;
    if url.scheme() != "https" || url.path() == "/" {
        return Err(OAuthClientError::BuildError(format!(
            "CIMD client ID `{client_id}` must use HTTPS and contain a path"
        )));
    }
    Ok(())
}

/// RFC 6749 §5.2 token error response, e.g. `{"error": "invalid_grant"}`.
#[derive(serde::Deserialize)]
struct TokenErrorResponse {
    error: String,
}

/// Return the RFC 6749 §5.2 `error` code from a token endpoint error
/// response, if the body parses as one.
fn token_error_code(response: &OAuthHttpResponse) -> Option<String> {
    serde_json::from_slice::<TokenErrorResponse>(&response.body)
        .ok()
        .map(|error| error.error)
}

async fn send_token_request(
    http: &dyn OAuthHttpClient,
    token_endpoint: &str,
    issuer: &str,
    registration: &OAuthClientRegistration,
    method: OAuthTokenEndpointAuthMethod,
    mut fields: Vec<(String, String)>,
    assertion_signer: Option<&dyn OAuthClientAssertionSigner>,
) -> Result<OAuthHttpResponse, OAuthClientError> {
    let is_refresh = fields
        .iter()
        .any(|(name, value)| name == "grant_type" && value == "refresh_token");
    let mut request = OAuthHttpRequest::post_form(token_endpoint, Vec::new());
    match method {
        OAuthTokenEndpointAuthMethod::None => {
            fields.push((
                "client_id".to_string(),
                registration.client_id().to_string(),
            ));
        }
        OAuthTokenEndpointAuthMethod::ClientSecretBasic => {
            let secret = registration.client_secret().ok_or_else(|| {
                OAuthClientError::BuildError(
                    "client_secret_basic selected without a client secret".to_string(),
                )
            })?;
            request = request.basic_auth(registration.client_id(), secret);
        }
        OAuthTokenEndpointAuthMethod::ClientSecretPost => {
            let secret = registration.client_secret().ok_or_else(|| {
                OAuthClientError::BuildError(
                    "client_secret_post selected without a client secret".to_string(),
                )
            })?;
            fields.push((
                "client_id".to_string(),
                registration.client_id().to_string(),
            ));
            fields.push(("client_secret".to_string(), secret.to_string()));
        }
        OAuthTokenEndpointAuthMethod::PrivateKeyJwt => {
            let signer = assertion_signer.ok_or_else(|| {
                OAuthClientError::BuildError(
                    "private_key_jwt selected without a client assertion signer".to_string(),
                )
            })?;
            let assertion = signer
                .sign_client_assertion(OAuthClientAssertionRequest {
                    client_id: registration.client_id().to_string(),
                    token_endpoint: token_endpoint.to_string(),
                    authorization_server_issuer: issuer.to_string(),
                })
                .await?;
            fields.push((
                "client_id".to_string(),
                registration.client_id().to_string(),
            ));
            fields.push((
                "client_assertion_type".to_string(),
                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer".to_string(),
            ));
            fields.push(("client_assertion".to_string(), assertion));
        }
    }
    request.body = OAuthHttpBody::Form(fields);
    let response = http.execute(request).await?;
    if !response.is_success() {
        // RFC 6749 §5.2: `invalid_grant` on a refresh means the refresh
        // token itself is dead (revoked or expired at the AS), not that
        // this particular request failed transiently. Surface that
        // distinctly so callers can discard the token instead of retrying
        // it forever.
        if is_refresh
            && response.status == 400
            && token_error_code(&response).as_deref() == Some("invalid_grant")
        {
            return Err(OAuthClientError::TokenRefreshRejected(
                response.error_body(),
            ));
        }
        return Err(OAuthClientError::TokenRequest(format!(
            "token endpoint returned HTTP {}: {}",
            response.status,
            response.error_body()
        )));
    }
    Ok(response)
}

fn token_from_response(
    response: OAuthHttpResponse,
    requested_scopes: &[String],
    previous_refresh_token: Option<String>,
) -> Result<OAuthStoredToken, OAuthClientError> {
    #[derive(serde::Deserialize)]
    struct TokenResponse {
        access_token: String,
        token_type: String,
        expires_in: Option<u64>,
        refresh_token: Option<String>,
        scope: Option<String>,
    }
    let response: TokenResponse = response.json()?;
    if !response.token_type.eq_ignore_ascii_case("bearer") {
        return Err(OAuthClientError::InvalidResponse(format!(
            "token endpoint returned unsupported token type `{}`",
            response.token_type
        )));
    }
    let scopes = response
        .scope
        .as_deref()
        .map(|scope| unique_scopes(scope.split_ascii_whitespace()))
        .unwrap_or_else(|| requested_scopes.to_vec());
    Ok(OAuthStoredToken {
        access_token: response.access_token,
        refresh_token: response.refresh_token.or(previous_refresh_token),
        expires_at: response
            .expires_in
            .map(|lifetime| unix_time().saturating_add(lifetime))
            .unwrap_or(u64::MAX),
        scopes,
    })
}

fn token_is_valid(token: &OAuthStoredToken, buffer: Duration) -> bool {
    unix_time().saturating_add(buffer.as_secs()) < token.expires_at
}

/// Whether `a` and `b` are different credentials rather than the same token
/// persisted twice.
///
/// The access token is what identifies a token issuance; a refresh always
/// mints a new one, so comparing it is enough to detect that another flow
/// rotated the token. `expires_at` must not decide this: `invalidate()`
/// (#1467) zeroes the in-memory token's `expires_at` without touching the
/// store or the refresh token, so comparing full structs (or the refresh
/// token, which `invalidate()` deliberately leaves untouched to reuse for
/// the coming refresh) would treat a reload of that same, still-unrevoked
/// store entry as a genuinely newer token and adopt back the very token
/// `invalidate()` just marked unusable.
fn tokens_differ(a: &OAuthStoredToken, b: &OAuthStoredToken) -> bool {
    a.access_token != b.access_token
}

fn scopes_are_covered(requested: &[String], granted: &[String]) -> bool {
    requested
        .iter()
        .all(|scope| granted.iter().any(|granted| granted == scope))
}

fn unix_time() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn unique_scopes<I, S>(scopes: I) -> Vec<String>
where
    I: IntoIterator<Item = S>,
    S: AsRef<str>,
{
    let mut unique = Vec::new();
    for scope in scopes {
        for scope in scope.as_ref().split_ascii_whitespace() {
            if !scope.is_empty() && !unique.iter().any(|existing| existing == scope) {
                unique.push(scope.to_string());
            }
        }
    }
    unique
}

fn random_urlsafe(bytes: usize) -> String {
    let mut value = vec![0_u8; bytes];
    getrandom::fill(&mut value).expect("getrandom failed");
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(value)
}

fn pkce_challenge(verifier: &str) -> String {
    use sha2::{Digest, Sha256};
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes()))
}

async fn prepare_redirect(
    policy: &OAuthRedirectPolicy,
    _state: &str,
) -> Result<(String, Option<oneshot::Receiver<String>>), OAuthClientError> {
    match policy {
        OAuthRedirectPolicy::Fixed { redirect_uri } => Ok((redirect_uri.clone(), None)),
        OAuthRedirectPolicy::Loopback {
            port,
            callback_path,
        } => {
            let listener = tokio::net::TcpListener::bind(("127.0.0.1", port.unwrap_or(0)))
                .await
                .map_err(|error| OAuthClientError::Redirect(error.to_string()))?;
            let actual_port = listener
                .local_addr()
                .map_err(|error| OAuthClientError::Redirect(error.to_string()))?
                .port();
            let redirect_uri = format!("http://127.0.0.1:{actual_port}{callback_path}");
            let (sender, receiver) = oneshot::channel();
            let callback_base = format!("http://127.0.0.1:{actual_port}");
            tokio::spawn(run_loopback_callback(listener, sender, callback_base));
            Ok((redirect_uri, Some(receiver)))
        }
    }
}

async fn run_loopback_callback(
    listener: tokio::net::TcpListener,
    mut sender: oneshot::Sender<String>,
    callback_base: String,
) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    let accepted = tokio::select! {
        _ = sender.closed() => return,
        accepted = listener.accept() => accepted,
    };
    let Ok((mut stream, _)) = accepted else {
        return;
    };
    let mut bytes = vec![0_u8; 8192];
    let Ok(read) = stream.read(&mut bytes).await else {
        return;
    };
    let request = String::from_utf8_lossy(&bytes[..read]);
    let target = request
        .lines()
        .next()
        .and_then(|line| line.split_ascii_whitespace().nth(1));
    let (status, body) = if target.is_some() {
        ("200 OK", "Authorization received. You can close this tab.")
    } else {
        ("400 Bad Request", "Invalid OAuth callback.")
    };
    let response = format!(
        "HTTP/1.1 {status}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    );
    let _ = stream.write_all(response.as_bytes()).await;
    let _ = stream.flush().await;
    if let Some(target) = target {
        let _ = sender.send(format!("{callback_base}{target}"));
    }
}

struct ParsedCallback {
    code: String,
    state: String,
    issuer: Option<String>,
}

fn parse_callback_url(callback_url: &str) -> Result<ParsedCallback, OAuthClientError> {
    let url = reqwest::Url::parse(callback_url)
        .map_err(|error| OAuthClientError::InvalidResponse(error.to_string()))?;
    let mut code = None;
    let mut state = None;
    let mut issuer = None;
    let mut error = None;
    let mut error_description = None;
    for (name, value) in url.query_pairs() {
        match name.as_ref() {
            "code" => code = Some(value.into_owned()),
            "state" => state = Some(value.into_owned()),
            "iss" => issuer = Some(value.into_owned()),
            "error" => error = Some(value.into_owned()),
            "error_description" => error_description = Some(value.into_owned()),
            _ => {}
        }
    }
    if let Some(error) = error {
        return Err(OAuthClientError::InvalidResponse(format!(
            "authorization server returned `{error}`{}",
            error_description
                .map(|description| format!(": {description}"))
                .unwrap_or_default()
        )));
    }
    Ok(ParsedCallback {
        code: code.ok_or_else(|| {
            OAuthClientError::InvalidResponse("callback omitted authorization code".to_string())
        })?,
        state: state.ok_or_else(|| {
            OAuthClientError::InvalidResponse("callback omitted state".to_string())
        })?,
        issuer,
    })
}

fn validate_callback_target(
    callback_url: &str,
    redirect_uri: &str,
) -> Result<(), OAuthClientError> {
    let callback = reqwest::Url::parse(callback_url)
        .map_err(|error| OAuthClientError::InvalidResponse(error.to_string()))?;
    let expected = reqwest::Url::parse(redirect_uri)
        .map_err(|error| OAuthClientError::InvalidResponse(error.to_string()))?;
    if callback.scheme() == expected.scheme()
        && callback.host_str() == expected.host_str()
        && callback.port_or_known_default() == expected.port_or_known_default()
        && callback.path() == expected.path()
    {
        Ok(())
    } else {
        Err(OAuthClientError::InvalidResponse(
            "callback URL does not match the configured redirect URI".to_string(),
        ))
    }
}

fn validate_callback_issuer(
    actual: Option<&str>,
    expected: &str,
    required: bool,
) -> Result<(), OAuthClientError> {
    match actual {
        Some(actual) if actual == expected => Ok(()),
        Some(actual) => Err(OAuthClientError::InvalidResponse(format!(
            "authorization response issuer mismatch: expected `{expected}`, got `{actual}`"
        ))),
        None if required => Err(OAuthClientError::InvalidResponse(
            "authorization response omitted required `iss`".to_string(),
        )),
        None => Ok(()),
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;
    use crate::client::OAuthScopeChallenge;

    #[derive(Debug, Clone, Copy)]
    enum RegistrationMode {
        PreRegistered,
        Cimd,
        Dynamic,
        /// DCR against an AS that only advertises `client_secret_basic`, whose
        /// registration endpoint returns no `client_secret`.
        DynamicConfidentialWithoutSecret,
        PrivateKeyJwt,
    }

    #[derive(Debug, Clone, Copy)]
    enum RefreshFailure {
        InvalidGrant,
        ServerError,
    }

    /// State for a mock authorization server that rotates the refresh token
    /// on every successful refresh (single use) and rejects a refresh token
    /// that has already been consumed with RFC 6749 `invalid_grant`. Used to
    /// simulate an AS that two flows sharing one token store can race
    /// (#1472).
    #[derive(Default)]
    struct RotatingRefresh {
        valid_refresh_token: String,
        generation: usize,
    }

    #[derive(Clone)]
    struct MockOAuthHttp {
        mode: RegistrationMode,
        requests: Arc<Mutex<Vec<OAuthHttpRequest>>>,
        token_requests: Arc<AtomicUsize>,
        expire_initial_token: bool,
        refresh_failure: Option<RefreshFailure>,
        rotating_refresh: Option<Arc<Mutex<RotatingRefresh>>>,
        no_resource_scopes: bool,
    }

    impl MockOAuthHttp {
        fn new(mode: RegistrationMode) -> Self {
            Self {
                mode,
                requests: Arc::new(Mutex::new(Vec::new())),
                token_requests: Arc::new(AtomicUsize::new(0)),
                expire_initial_token: false,
                refresh_failure: None,
                rotating_refresh: None,
                no_resource_scopes: false,
            }
        }

        /// The 401 challenge carries no `scope` and the protected-resource
        /// metadata has no `scopes_supported`, while the authorization server
        /// still advertises scopes of its own.
        fn without_resource_scopes(mut self) -> Self {
            self.no_resource_scopes = true;
            self
        }

        fn expiring(mut self) -> Self {
            self.expire_initial_token = true;
            self
        }

        /// The token endpoint rejects a refresh with RFC 6749 `invalid_grant`.
        fn refresh_rejected(mut self) -> Self {
            self.refresh_failure = Some(RefreshFailure::InvalidGrant);
            self
        }

        /// The token endpoint fails a refresh with a transient HTTP 500.
        fn refresh_server_error(mut self) -> Self {
            self.refresh_failure = Some(RefreshFailure::ServerError);
            self
        }

        /// The token endpoint issues a single-use refresh token and rotates
        /// it on every successful refresh, rejecting reuse of a consumed one
        /// with `invalid_grant`.
        fn rotating_refresh_tokens(mut self) -> Self {
            self.rotating_refresh = Some(Arc::new(Mutex::new(RotatingRefresh {
                valid_refresh_token: "refresh-token-0".to_string(),
                generation: 0,
            })));
            self
        }

        async fn requests(&self) -> Vec<OAuthHttpRequest> {
            self.requests.lock().await.clone()
        }

        fn response(status: u16, body: serde_json::Value) -> OAuthHttpResponse {
            OAuthHttpResponse {
                status,
                headers: Vec::new(),
                body: serde_json::to_vec(&body).unwrap(),
            }
        }
    }

    #[async_trait]
    impl OAuthHttpClient for MockOAuthHttp {
        async fn execute(
            &self,
            request: OAuthHttpRequest,
        ) -> Result<OAuthHttpResponse, OAuthClientError> {
            self.requests.lock().await.push(request.clone());
            let path = reqwest::Url::parse(&request.url)
                .unwrap()
                .path()
                .to_string();
            match path.as_str() {
                "/mcp" => Ok(OAuthHttpResponse {
                    status: 401,
                    headers: vec![(
                        "www-authenticate".to_string(),
                        if self.no_resource_scopes {
                            "Bearer resource_metadata=\"https://mcp.example.com/prm\"".to_string()
                        } else {
                            "Bearer resource_metadata=\"https://mcp.example.com/prm\", scope=\"challenge.scope\""
                                .to_string()
                        },
                    )],
                    body: Vec::new(),
                }),
                "/prm" => Ok(Self::response(
                    200,
                    if self.no_resource_scopes {
                        serde_json::json!({
                            "resource": "https://mcp.example.com/mcp",
                            "authorization_servers": ["https://auth.example.com/issuer"]
                        })
                    } else {
                        serde_json::json!({
                            "resource": "https://mcp.example.com/mcp",
                            "authorization_servers": ["https://auth.example.com/issuer"],
                            "scopes_supported": ["prm.scope"]
                        })
                    },
                )),
                "/.well-known/oauth-authorization-server/issuer" => {
                    let (cimd, methods) = match self.mode {
                        RegistrationMode::PreRegistered => (false, vec!["client_secret_basic"]),
                        RegistrationMode::Cimd => (true, vec!["none"]),
                        RegistrationMode::Dynamic => (false, vec!["none"]),
                        RegistrationMode::DynamicConfidentialWithoutSecret => {
                            (false, vec!["client_secret_basic"])
                        }
                        RegistrationMode::PrivateKeyJwt => (false, vec!["private_key_jwt"]),
                    };
                    let as_scopes = if self.no_resource_scopes {
                        vec!["as.scope", "other.scope"]
                    } else {
                        vec![
                            "challenge.scope",
                            "prm.scope",
                            "extra.scope",
                            "offline_access",
                        ]
                    };
                    Ok(Self::response(
                        200,
                        serde_json::json!({
                            "issuer": "https://auth.example.com/issuer",
                            "authorization_endpoint": "https://auth.example.com/authorize",
                            "token_endpoint": "https://auth.example.com/token",
                            "registration_endpoint": "https://auth.example.com/register",
                            "client_id_metadata_document_supported": cimd,
                            "authorization_response_iss_parameter_supported": true,
                            "code_challenge_methods_supported": ["S256"],
                            "token_endpoint_auth_methods_supported": methods,
                            "grant_types_supported": ["authorization_code", "refresh_token"],
                            "scopes_supported": as_scopes
                        }),
                    ))
                }
                "/register" => Ok(Self::response(
                    201,
                    serde_json::json!({ "client_id": "dynamic-client" }),
                )),
                "/token" => {
                    let request_number = self.token_requests.fetch_add(1, Ordering::SeqCst);
                    let fields = match &request.body {
                        OAuthHttpBody::Form(fields) => fields,
                        body => panic!("expected form token request, got {body:?}"),
                    };
                    let grant = fields
                        .iter()
                        .find(|(name, _)| name == "grant_type")
                        .map(|(_, value)| value.as_str())
                        .unwrap();
                    let scope = fields
                        .iter()
                        .find(|(name, _)| name == "scope")
                        .map(|(_, value)| value.clone())
                        .unwrap_or_else(|| {
                            if request_number == 0 {
                                "challenge.scope offline_access".to_string()
                            } else {
                                "challenge.scope offline_access extra.scope".to_string()
                            }
                        });
                    if grant == "refresh_token" {
                        if let Some(state) = &self.rotating_refresh {
                            let refresh_token = fields
                                .iter()
                                .find(|(name, _)| name == "refresh_token")
                                .map(|(_, value)| value.as_str())
                                .unwrap();
                            let mut state = state.lock().await;
                            return if refresh_token == state.valid_refresh_token {
                                state.generation += 1;
                                let new_refresh_token =
                                    format!("refresh-token-{}", state.generation);
                                state.valid_refresh_token = new_refresh_token.clone();
                                Ok(Self::response(
                                    200,
                                    serde_json::json!({
                                        "access_token": format!(
                                            "rotated-access-token-{}",
                                            state.generation
                                        ),
                                        "token_type": "Bearer",
                                        "expires_in": 3600,
                                        "refresh_token": new_refresh_token,
                                        "scope": scope
                                    }),
                                ))
                            } else {
                                // This refresh token was already consumed by
                                // an earlier rotation; a real AS answers
                                // reuse of a single-use refresh token with
                                // invalid_grant.
                                Ok(Self::response(
                                    400,
                                    serde_json::json!({
                                        "error": "invalid_grant",
                                        "error_description": "refresh token already used"
                                    }),
                                ))
                            };
                        }
                        match self.refresh_failure {
                            Some(RefreshFailure::InvalidGrant) => Ok(Self::response(
                                400,
                                serde_json::json!({
                                    "error": "invalid_grant",
                                    "error_description": "refresh token expired or revoked"
                                }),
                            )),
                            Some(RefreshFailure::ServerError) => Ok(OAuthHttpResponse {
                                status: 500,
                                headers: Vec::new(),
                                body: b"internal server error".to_vec(),
                            }),
                            None => Ok(Self::response(
                                200,
                                serde_json::json!({
                                    "access_token": "refreshed-token",
                                    "token_type": "Bearer",
                                    "expires_in": 3600,
                                    "scope": scope
                                }),
                            )),
                        }
                    } else {
                        let refresh_token = match &self.rotating_refresh {
                            Some(state) => state.lock().await.valid_refresh_token.clone(),
                            None => "refresh-token".to_string(),
                        };
                        Ok(Self::response(
                            200,
                            serde_json::json!({
                                "access_token": format!("access-token-{request_number}"),
                                "token_type": "Bearer",
                                "expires_in": if self.expire_initial_token { 0 } else { 3600 },
                                "refresh_token": refresh_token,
                                "scope": scope
                            }),
                        ))
                    }
                }
                other => Err(OAuthClientError::Http(format!(
                    "unexpected mock request path `{other}`"
                ))),
            }
        }
    }

    #[derive(Clone, Default)]
    struct AutomaticAuthorizationHandler {
        calls: Arc<AtomicUsize>,
    }

    #[derive(Clone, Default)]
    struct TestAssertionSigner;

    #[async_trait]
    impl OAuthClientAssertionSigner for TestAssertionSigner {
        async fn sign_client_assertion(
            &self,
            request: OAuthClientAssertionRequest,
        ) -> Result<String, OAuthClientError> {
            assert_eq!(request.client_id, "signed-client");
            assert_eq!(request.token_endpoint, "https://auth.example.com/token");
            assert_eq!(
                request.authorization_server_issuer,
                "https://auth.example.com/issuer"
            );
            Ok("signed-client-assertion".to_string())
        }
    }

    #[async_trait]
    impl OAuthAuthorizationHandler for AutomaticAuthorizationHandler {
        async fn authorize(
            &self,
            request: OAuthAuthorizationRequest,
        ) -> Result<OAuthAuthorizationAction, OAuthClientError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            let authorization_url = reqwest::Url::parse(&request.authorization_url).unwrap();
            let state = authorization_url
                .query_pairs()
                .find(|(name, _)| name == "state")
                .unwrap()
                .1
                .into_owned();
            let mut callback = reqwest::Url::parse(&request.redirect_uri).unwrap();
            callback
                .query_pairs_mut()
                .append_pair("code", "authorization-code")
                .append_pair("state", &state)
                .append_pair("iss", &request.issuer);
            Ok(OAuthAuthorizationAction::CallbackUrl(callback.to_string()))
        }
    }

    fn dynamic_options() -> OAuthClientRegistrationOptions {
        OAuthClientRegistrationOptions::new()
            .with_client_id_metadata_document("https://client.example.com/metadata.json")
            .with_dynamic_registration(
                super::super::oauth_authcode::OAuthDynamicClientRegistration::native(
                    "test-client",
                    std::iter::empty::<String>(),
                ),
            )
    }

    fn flow_builder(
        http: MockOAuthHttp,
        options: OAuthClientRegistrationOptions,
        handler: AutomaticAuthorizationHandler,
    ) -> OAuthAuthorizationFlowBuilder {
        OAuthAuthorizationFlow::builder("https://mcp.example.com/mcp")
            .http_client(http)
            .redirect_policy(OAuthRedirectPolicy::fixed(
                "http://127.0.0.1:23456/callback",
            ))
            .registration_options(options)
            .authorization_handler(handler)
    }

    #[tokio::test]
    async fn preregistered_flow_binds_resource_and_uses_basic_auth() {
        let http = MockOAuthHttp::new(RegistrationMode::PreRegistered);
        let flow = flow_builder(
            http.clone(),
            OAuthClientRegistrationOptions::new(),
            AutomaticAuthorizationHandler::default(),
        )
        .pre_registered_client("pre:client", Some("pre secret".to_string()))
        .build()
        .unwrap();

        flow.authorize(std::iter::empty::<&str>()).await.unwrap();
        assert_eq!(flow.get_token().await.unwrap(), "access-token-0");
        assert_eq!(
            flow.authorized_scopes().await.unwrap(),
            vec!["challenge.scope", "offline_access"]
        );

        let requests = http.requests().await;
        let probe = requests.first().unwrap();
        assert_eq!(probe.method, OAuthHttpMethod::Post);
        let OAuthHttpBody::Json(probe_body) = &probe.body else {
            panic!("expected JSON MCP probe")
        };
        assert_eq!(probe_body["method"], "tools/list");
        assert!(
            !requests
                .iter()
                .any(|request| request.url.ends_with("/register"))
        );
        let token_request = requests
            .iter()
            .find(|request| request.url.ends_with("/token"))
            .unwrap();
        assert!(token_request.headers.iter().any(|(name, value)| {
            name == "authorization"
                && value
                    == &format!(
                        "Basic {}",
                        base64::engine::general_purpose::STANDARD
                            .encode("pre%3Aclient:pre%20secret")
                    )
        }));
        let OAuthHttpBody::Form(fields) = &token_request.body else {
            panic!("expected token form")
        };
        assert!(fields.iter().any(|field| field
            == &(
                "resource".to_string(),
                "https://mcp.example.com/mcp".to_string()
            )));
    }

    #[tokio::test]
    async fn explicit_resource_override_used_for_authorize_exchange_and_refresh() {
        let http = MockOAuthHttp::new(RegistrationMode::PreRegistered).expiring();
        let override_resource = "https://override.example.com/aud".to_string();
        let flow = flow_builder(
            http.clone(),
            OAuthClientRegistrationOptions::new(),
            AutomaticAuthorizationHandler::default(),
        )
        .pre_registered_client("pre:client", Some("pre secret".to_string()))
        .resource(override_resource.clone())
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();

        let OAuthAuthorizationStart::Pending(pending) =
            flow.begin(std::iter::empty::<&str>()).await.unwrap()
        else {
            panic!("expected pending flow")
        };
        // The authorization request carries the explicit override, not the
        // resource discovered from PRM (https://mcp.example.com/mcp).
        assert_eq!(pending.request().resource, override_resource);
        let authorization_url = reqwest::Url::parse(&pending.request().authorization_url).unwrap();
        let parameters: HashMap<_, _> = authorization_url.query_pairs().into_owned().collect();
        assert_eq!(parameters.get("resource"), Some(&override_resource));
        let state = parameters.get("state").unwrap().clone();

        let request = pending.request().clone();
        let mut callback = reqwest::Url::parse(&request.redirect_uri).unwrap();
        callback
            .query_pairs_mut()
            .append_pair("code", "authorization-code")
            .append_pair("state", &state)
            .append_pair("iss", &request.issuer);
        pending
            .complete_callback_url(callback.as_str())
            .await
            .unwrap();

        // Code exchange carried the override.
        let requests = http.requests().await;
        let exchange = requests
            .iter()
            .find(|request| request.url.ends_with("/token"))
            .unwrap();
        let OAuthHttpBody::Form(fields) = &exchange.body else {
            panic!("expected token form")
        };
        assert!(
            fields
                .iter()
                .any(|field| field == &("resource".to_string(), override_resource.clone()))
        );

        // The immediately-expired token forces a refresh on the next call;
        // it must carry the override too.
        assert_eq!(flow.get_token().await.unwrap(), "refreshed-token");
        let requests = http.requests().await;
        let refresh = requests
            .iter()
            .rfind(|request| request.url.ends_with("/token"))
            .unwrap();
        let OAuthHttpBody::Form(fields) = &refresh.body else {
            panic!("expected refresh form")
        };
        assert!(
            fields
                .iter()
                .any(|field| field == &("grant_type".to_string(), "refresh_token".to_string()))
        );
        assert!(
            fields
                .iter()
                .any(|field| field == &("resource".to_string(), override_resource.clone()))
        );
    }

    #[tokio::test]
    async fn explicit_resource_override_changes_the_token_binding() {
        let http = MockOAuthHttp::new(RegistrationMode::PreRegistered);
        let override_resource = "https://override.example.com/aud".to_string();
        let tokens = MemoryOAuthTokenStore::new();

        // Seed a valid token under the binding the *discovered* resource
        // would produce.
        let discovered_binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "pre:client".to_string(),
        };
        let stale_token = OAuthStoredToken {
            access_token: "stale-discovered-token".to_string(),
            refresh_token: None,
            expires_at: u64::MAX,
            scopes: vec![],
        };
        tokens
            .save(&discovered_binding, &stale_token)
            .await
            .unwrap();

        let flow = flow_builder(
            http.clone(),
            OAuthClientRegistrationOptions::new(),
            AutomaticAuthorizationHandler::default(),
        )
        .pre_registered_client("pre:client", Some("pre secret".to_string()))
        .resource(override_resource.clone())
        .token_store(tokens.clone())
        .build()
        .unwrap();

        // Nothing is stored under the override binding yet, so the token
        // stored under the discovered resource must not be adopted.
        let OAuthAuthorizationStart::Pending(pending) =
            flow.begin(std::iter::empty::<&str>()).await.unwrap()
        else {
            panic!("a token stored under the discovered resource must not be reused")
        };

        let request = pending.request().clone();
        let state = reqwest::Url::parse(&request.authorization_url)
            .unwrap()
            .query_pairs()
            .find(|(name, _)| name == "state")
            .unwrap()
            .1
            .into_owned();
        let mut callback = reqwest::Url::parse(&request.redirect_uri).unwrap();
        callback
            .query_pairs_mut()
            .append_pair("code", "authorization-code")
            .append_pair("state", &state)
            .append_pair("iss", &request.issuer);
        pending
            .complete_callback_url(callback.as_str())
            .await
            .unwrap();

        let override_binding = OAuthTokenBinding {
            resource: override_resource,
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "pre:client".to_string(),
        };
        let stored = tokens.load(&override_binding).await.unwrap().unwrap();
        assert_eq!(stored.access_token, "access-token-0");

        // The entry under the discovered resource is untouched.
        let untouched = tokens.load(&discovered_binding).await.unwrap().unwrap();
        assert_eq!(untouched.access_token, "stale-discovered-token");
    }

    #[test]
    fn explicit_resource_override_with_fragment_is_rejected() {
        let err = OAuthAuthorizationFlow::builder("https://mcp.example.com/mcp")
            .redirect_policy(OAuthRedirectPolicy::fixed(
                "http://127.0.0.1:23456/callback",
            ))
            .resource("https://override.example.com/aud#frag")
            .build()
            .unwrap_err();
        assert!(matches!(err, OAuthClientError::BuildError(_)));
        assert!(err.to_string().contains("fragment"));
    }

    #[tokio::test]
    async fn private_key_jwt_is_used_when_advertised() {
        let http = MockOAuthHttp::new(RegistrationMode::PrivateKeyJwt);
        let options = OAuthClientRegistrationOptions::new().with_pre_registered(
            OAuthClientRegistration::pre_registered(
                "https://auth.example.com/issuer",
                "signed-client",
                None,
            ),
        );
        let flow = flow_builder(
            http.clone(),
            options,
            AutomaticAuthorizationHandler::default(),
        )
        .client_assertion_signer(TestAssertionSigner)
        .build()
        .unwrap();

        flow.authorize(["challenge.scope"]).await.unwrap();
        let requests = http.requests().await;
        let token_request = requests
            .iter()
            .find(|request| request.url.ends_with("/token"))
            .unwrap();
        let OAuthHttpBody::Form(fields) = &token_request.body else {
            panic!("expected token form")
        };
        assert!(fields.iter().any(|field| field
            == &(
                "client_assertion_type".to_string(),
                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer".to_string()
            )));
        assert!(fields.iter().any(|field| field
            == &(
                "client_assertion".to_string(),
                "signed-client-assertion".to_string()
            )));
    }

    #[tokio::test]
    async fn cimd_takes_priority_over_dynamic_registration() {
        let http = MockOAuthHttp::new(RegistrationMode::Cimd);
        let flow = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .build()
        .unwrap();

        flow.authorize(["explicit.scope"]).await.unwrap();
        let requests = http.requests().await;
        assert!(
            !requests
                .iter()
                .any(|request| request.url.ends_with("/register"))
        );
        let authorization = requests
            .iter()
            .find(|request| request.url.ends_with("/token"))
            .unwrap();
        let OAuthHttpBody::Form(fields) = &authorization.body else {
            panic!("expected form")
        };
        assert!(fields.iter().any(|field| field
            == &(
                "client_id".to_string(),
                "https://client.example.com/metadata.json".to_string()
            )));
        assert!(fields.iter().any(|field| field
            == &(
                "resource".to_string(),
                "https://mcp.example.com/mcp".to_string()
            )));
    }

    #[tokio::test]
    async fn dynamic_registration_is_persisted_and_reused() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic);
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();

        for _ in 0..2 {
            let flow = flow_builder(
                http.clone(),
                dynamic_options(),
                AutomaticAuthorizationHandler::default(),
            )
            .registration_store(registrations.clone())
            .build()
            .unwrap();
            flow.authorize(["challenge.scope"]).await.unwrap();
        }

        let requests = http.requests().await;
        assert_eq!(
            requests
                .iter()
                .filter(|request| request.url.ends_with("/register"))
                .count(),
            1
        );
        let registration_request = requests
            .iter()
            .find(|request| request.url.ends_with("/register"))
            .unwrap();
        let OAuthHttpBody::Json(value) = &registration_request.body else {
            panic!("expected registration JSON")
        };
        assert_eq!(value["redirect_uris"][0], "http://127.0.0.1:23456/callback");
    }

    /// #1558 (rmcp #1355): a confidential registration that comes back
    /// without a secret fails at registration, not later at the token
    /// endpoint after silently falling back to `none`.
    #[tokio::test]
    async fn confidential_dynamic_registration_without_secret_is_rejected() {
        let http = MockOAuthHttp::new(RegistrationMode::DynamicConfidentialWithoutSecret);
        let flow = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .build()
        .unwrap();

        let error = flow.authorize(["challenge.scope"]).await.unwrap_err();

        assert!(
            matches!(&error, OAuthClientError::Registration(message)
                if message.contains("client_secret_basic") && message.contains("client_secret")),
            "unexpected error: {error}"
        );
        let requests = http.requests().await;
        let registration_request = requests
            .iter()
            .find(|request| request.url.ends_with("/register"))
            .unwrap();
        let OAuthHttpBody::Json(value) = &registration_request.body else {
            panic!("expected registration JSON")
        };
        assert_eq!(value["token_endpoint_auth_method"], "client_secret_basic");
        assert!(
            !requests
                .iter()
                .any(|request| request.url.ends_with("/token")),
            "no token request should follow a failed registration"
        );
    }

    /// #1558 (rmcp PR #1352): with no challenge scope and no PRM
    /// `scopes_supported`, the MCP scope selection strategy omits `scope`
    /// rather than requesting everything the AS advertises.
    #[tokio::test]
    async fn authorization_url_omits_scope_without_challenge_or_resource_scopes() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).without_resource_scopes();
        let flow = flow_builder(
            http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .build()
        .unwrap();

        let OAuthAuthorizationStart::Pending(pending) =
            flow.begin(std::iter::empty::<&str>()).await.unwrap()
        else {
            panic!("expected pending flow")
        };

        let url = reqwest::Url::parse(&pending.request().authorization_url).unwrap();
        let parameters: HashMap<_, _> = url.query_pairs().into_owned().collect();
        assert!(
            !parameters.contains_key("scope"),
            "the authorization URL must not carry a scope"
        );
        assert!(pending.request().scopes.is_empty());

        // Explicit caller scopes still win.
        let OAuthAuthorizationStart::Pending(pending) =
            flow.begin(["explicit.scope"]).await.unwrap()
        else {
            panic!("expected pending flow")
        };
        let url = reqwest::Url::parse(&pending.request().authorization_url).unwrap();
        let parameters: HashMap<_, _> = url.query_pairs().into_owned().collect();
        assert_eq!(
            parameters.get("scope").map(String::as_str),
            Some("explicit.scope")
        );
    }

    #[tokio::test]
    async fn expired_token_refreshes_and_preserves_binding() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).expiring();
        let flow = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();

        flow.authorize(["challenge.scope"]).await.unwrap();
        assert_eq!(flow.get_token().await.unwrap(), "refreshed-token");
        let requests = http.requests().await;
        let refresh = requests
            .iter()
            .rfind(|request| request.url.ends_with("/token"))
            .unwrap();
        let OAuthHttpBody::Form(fields) = &refresh.body else {
            panic!("expected refresh form")
        };
        assert!(
            fields
                .iter()
                .any(|field| field == &("grant_type".to_string(), "refresh_token".to_string()))
        );
        assert!(fields.iter().any(|field| field
            == &(
                "resource".to_string(),
                "https://mcp.example.com/mcp".to_string()
            )));
    }

    #[tokio::test]
    async fn expired_persisted_token_refreshes_after_flow_rebuild() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).expiring();
        let tokens = MemoryOAuthTokenStore::new();
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();
        let first_handler = AutomaticAuthorizationHandler::default();

        let first = flow_builder(http.clone(), dynamic_options(), first_handler)
            .registration_store(registrations.clone())
            .token_store(tokens.clone())
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();
        first.authorize(["challenge.scope"]).await.unwrap();

        let restored_handler = AutomaticAuthorizationHandler::default();
        let restored_calls = restored_handler.calls.clone();
        let restored = flow_builder(http, dynamic_options(), restored_handler)
            .registration_store(registrations)
            .token_store(tokens)
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();

        let start = restored.begin(["challenge.scope"]).await.unwrap();
        assert!(matches!(start, OAuthAuthorizationStart::Authorized { .. }));
        assert_eq!(restored_calls.load(Ordering::SeqCst), 0);
        assert_eq!(restored.get_token().await.unwrap(), "refreshed-token");
    }

    #[tokio::test]
    async fn rejected_refresh_of_persisted_token_reauthorizes_and_discards_it() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).expiring();
        let tokens = MemoryOAuthTokenStore::new();
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();
        let first_handler = AutomaticAuthorizationHandler::default();

        let first = flow_builder(http.clone(), dynamic_options(), first_handler)
            .registration_store(registrations.clone())
            .token_store(tokens.clone())
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();
        first.authorize(["challenge.scope"]).await.unwrap();

        let binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "dynamic-client".to_string(),
        };
        assert!(tokens.load(&binding).await.unwrap().is_some());

        // Rebuild the flow against an AS that now rejects the refresh token
        // (revoked or expired since the token was persisted).
        let rejecting_http = MockOAuthHttp::new(RegistrationMode::Dynamic).refresh_rejected();
        let restored_handler = AutomaticAuthorizationHandler::default();
        let restored_calls = restored_handler.calls.clone();
        let restored = flow_builder(rejecting_http, dynamic_options(), restored_handler)
            .registration_store(registrations)
            .token_store(tokens.clone())
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();

        let start = restored.begin(["challenge.scope"]).await.unwrap();
        assert!(matches!(start, OAuthAuthorizationStart::Pending(_)));
        // begin() only returns a pending request; it never drives the
        // authorization handler itself.
        assert_eq!(restored_calls.load(Ordering::SeqCst), 0);
        assert!(tokens.load(&binding).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn transient_refresh_failure_keeps_persisted_token() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).expiring();
        let tokens = MemoryOAuthTokenStore::new();
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();
        let first_handler = AutomaticAuthorizationHandler::default();

        let first = flow_builder(http.clone(), dynamic_options(), first_handler)
            .registration_store(registrations.clone())
            .token_store(tokens.clone())
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();
        first.authorize(["challenge.scope"]).await.unwrap();

        let binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "dynamic-client".to_string(),
        };
        assert!(tokens.load(&binding).await.unwrap().is_some());

        // A transient failure (HTTP 500, not RFC 6749 invalid_grant) must
        // propagate as an error and must not discard the persisted token.
        let failing_http = MockOAuthHttp::new(RegistrationMode::Dynamic).refresh_server_error();
        let restored = flow_builder(
            failing_http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .registration_store(registrations)
        .token_store(tokens.clone())
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();

        let error = restored.begin(["challenge.scope"]).await.unwrap_err();
        assert!(matches!(error, OAuthClientError::TokenRequest(_)));
        assert!(tokens.load(&binding).await.unwrap().is_some());
    }

    /// #1467: `invalidate` previously inherited the no-op default, so a 401
    /// on a token that had not yet expired retried the same rejected token
    /// forever. It must mark the token unusable so the next `get_token`
    /// refreshes even though nothing about expiry has changed.
    #[tokio::test]
    async fn invalidate_forces_a_refresh_of_a_still_valid_token() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic);
        let flow = flow_builder(
            http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .build()
        .unwrap();
        flow.authorize(["challenge.scope"]).await.unwrap();
        assert_eq!(flow.get_token().await.unwrap(), "access-token-0");

        flow.invalidate().await;

        assert_eq!(
            flow.get_token().await.unwrap(),
            "refreshed-token",
            "invalidate() must mark the access token unusable so the next \
             get_token() refreshes instead of returning the same token"
        );
    }

    /// #1467: with no refresh token to fall back on, `invalidate` must leave
    /// `get_token` reporting that reauthorization is needed rather than
    /// silently handing back the same now-unusable token.
    #[tokio::test]
    async fn invalidate_without_a_refresh_token_reports_reauthorization_is_needed() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic);
        let flow = flow_builder(
            http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .build()
        .unwrap();
        flow.authorize(["challenge.scope"]).await.unwrap();

        // Simulate a token issued without a refresh token (no offline_access
        // grant), which the mock token endpoint above does not produce.
        {
            let mut current = flow.inner.current.write().await;
            current.as_mut().unwrap().token.refresh_token = None;
        }

        flow.invalidate().await;

        let error = flow.get_token().await.unwrap_err();
        let message = error.to_string();
        assert!(
            message.contains("reauthorize"),
            "expected the error to say reauthorization is needed, got: {message}"
        );
    }

    /// #1472: two flows share one token store and the authorization server
    /// rotates refresh tokens on use (single-use). Flow A refreshes first,
    /// rotating R0 -> R1 and saving R1 to the shared store. Flow B still
    /// holds R0 in memory with an expired access token; if it refreshed with
    /// R0 the AS would already have consumed it and answer invalid_grant.
    /// Flow B must instead notice the store already holds a newer, valid
    /// token and adopt it rather than failing or requiring reauthorization.
    #[tokio::test]
    async fn two_flows_sharing_a_store_survive_a_rotated_refresh_token() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic)
            .expiring()
            .rotating_refresh_tokens();
        let tokens = MemoryOAuthTokenStore::new();
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();

        let flow_a = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .registration_store(registrations.clone())
        .token_store(tokens.clone())
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();
        flow_a.authorize(["challenge.scope"]).await.unwrap();

        let flow_b = flow_builder(
            http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .registration_store(registrations)
        .token_store(tokens.clone())
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();
        // Give flow B the same starting token flow A had, as if it had
        // loaded it from the shared store before flow A's refresh ran.
        {
            let active = flow_a.inner.current.read().await.clone().unwrap();
            *flow_b.inner.current.write().await = Some(active);
        }

        // Flow A refreshes first: R0 -> R1, persisted to the shared store.
        assert_eq!(flow_a.get_token().await.unwrap(), "rotated-access-token-1");

        // Flow B still holds R0, which the AS has already consumed. It must
        // end up with a valid token instead of failing.
        let token = flow_b
            .get_token()
            .await
            .expect("flow B must recover instead of failing when another flow already rotated the refresh token");
        assert_eq!(token, "rotated-access-token-1");

        let binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "dynamic-client".to_string(),
        };
        let stored = tokens.load(&binding).await.unwrap().unwrap();
        assert!(token_is_valid(&stored, Duration::ZERO));
    }

    /// #1472: when the shared store already holds a different, valid token
    /// for this binding, `get_token` must adopt it instead of spending a
    /// network round trip on a refresh.
    #[tokio::test]
    async fn get_token_adopts_a_newer_stored_token_without_refreshing() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic).expiring();
        let tokens = MemoryOAuthTokenStore::new();
        let flow = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .token_store(tokens.clone())
        .refresh_buffer(Duration::ZERO)
        .build()
        .unwrap();
        flow.authorize(["challenge.scope"]).await.unwrap();
        let requests_before = http.token_requests.load(Ordering::SeqCst);

        let binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "dynamic-client".to_string(),
        };
        let newer = OAuthStoredToken {
            access_token: "externally-refreshed-token".to_string(),
            refresh_token: Some("externally-refreshed-refresh-token".to_string()),
            expires_at: unix_time() + 3600,
            scopes: vec!["challenge.scope".to_string()],
        };
        tokens.save(&binding, &newer).await.unwrap();

        let token = flow.get_token().await.unwrap();
        assert_eq!(token, "externally-refreshed-token");
        assert_eq!(
            http.token_requests.load(Ordering::SeqCst),
            requests_before,
            "adopting a newer stored token must not perform a refresh request"
        );
    }

    /// #1472: an authorization server can reject a refresh with
    /// invalid_grant for a reason unrelated to permanent revocation: another
    /// flow sharing this store consumed the refresh token an instant before
    /// this request reached the server. Reloading the store once after the
    /// rejection, before discarding, must find that newer token and adopt
    /// it instead of forcing reauthorization.
    #[tokio::test]
    async fn invalid_grant_recovers_when_store_gained_a_newer_token_mid_refresh() {
        struct RaceOnRefreshHttp {
            inner: MockOAuthHttp,
            tokens: MemoryOAuthTokenStore,
            binding: OAuthTokenBinding,
            winner: OAuthStoredToken,
            injected: Arc<AtomicUsize>,
        }

        #[async_trait]
        impl OAuthHttpClient for RaceOnRefreshHttp {
            async fn execute(
                &self,
                request: OAuthHttpRequest,
            ) -> Result<OAuthHttpResponse, OAuthClientError> {
                let is_refresh_request = request.url.ends_with("/token")
                    && matches!(
                        &request.body,
                        OAuthHttpBody::Form(fields)
                            if fields.iter().any(|(name, value)| name == "grant_type" && value == "refresh_token")
                    );
                if is_refresh_request && self.injected.fetch_add(1, Ordering::SeqCst) == 0 {
                    // Simulate another flow's rotation landing in the window
                    // between this flow's pre-refresh store reload and the
                    // authorization server actually answering this request.
                    self.tokens.save(&self.binding, &self.winner).await.unwrap();
                }
                self.inner.execute(request).await
            }
        }

        let tokens = MemoryOAuthTokenStore::new();
        let binding = OAuthTokenBinding {
            resource: "https://mcp.example.com/mcp".to_string(),
            issuer: "https://auth.example.com/issuer".to_string(),
            client_id: "dynamic-client".to_string(),
        };
        let winner = OAuthStoredToken {
            access_token: "winner-access-token".to_string(),
            refresh_token: Some("winner-refresh-token".to_string()),
            expires_at: unix_time() + 3600,
            scopes: vec!["challenge.scope".to_string()],
        };
        let http = RaceOnRefreshHttp {
            inner: MockOAuthHttp::new(RegistrationMode::Dynamic)
                .expiring()
                .refresh_rejected(),
            tokens: tokens.clone(),
            binding: binding.clone(),
            winner: winner.clone(),
            injected: Arc::new(AtomicUsize::new(0)),
        };
        let flow = OAuthAuthorizationFlow::builder("https://mcp.example.com/mcp")
            .http_client(http)
            .redirect_policy(OAuthRedirectPolicy::fixed(
                "http://127.0.0.1:23456/callback",
            ))
            .registration_options(dynamic_options())
            .authorization_handler(AutomaticAuthorizationHandler::default())
            .token_store(tokens.clone())
            .refresh_buffer(Duration::ZERO)
            .build()
            .unwrap();
        flow.authorize(["challenge.scope"]).await.unwrap();

        let token = flow
            .get_token()
            .await
            .expect("must adopt the winner's token instead of failing on invalid_grant");
        assert_eq!(token, winner.access_token);
        let stored = tokens.load(&binding).await.unwrap().unwrap();
        assert_eq!(stored.access_token, winner.access_token);
    }

    #[tokio::test]
    async fn dropped_loopback_attempt_releases_its_listener() {
        let probe = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .unwrap();
        let port = probe.local_addr().unwrap().port();
        drop(probe);

        let (_, receiver) = prepare_redirect(
            &OAuthRedirectPolicy::loopback_at(port, "/callback"),
            "state",
        )
        .await
        .unwrap();
        drop(receiver);

        tokio::time::timeout(Duration::from_secs(1), async {
            loop {
                match tokio::net::TcpListener::bind(("127.0.0.1", port)).await {
                    Ok(listener) => break drop(listener),
                    Err(_) => tokio::task::yield_now().await,
                }
            }
        })
        .await
        .expect("dropped OAuth attempt kept its loopback port bound");
    }

    #[tokio::test]
    async fn scope_escalation_reauthorizes_same_provider() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic);
        let handler = AutomaticAuthorizationHandler::default();
        let calls = handler.calls.clone();
        let flow = flow_builder(http, dynamic_options(), handler)
            .build()
            .unwrap();
        flow.authorize(std::iter::empty::<&str>()).await.unwrap();

        flow.reauthorize(OAuthScopeEscalationRequest {
            resource: "https://mcp.example.com/mcp".to_string(),
            operation: "tools/call:admin".to_string(),
            challenge: OAuthScopeChallenge {
                required_scopes: vec!["extra.scope".to_string()],
                resource_metadata: Some("https://mcp.example.com/prm".to_string()),
                error_description: None,
            },
            previous_scopes: vec!["challenge.scope".to_string(), "offline_access".to_string()],
            requested_scopes: vec![
                "challenge.scope".to_string(),
                "offline_access".to_string(),
                "extra.scope".to_string(),
            ],
            attempt: 1,
        })
        .await
        .unwrap();

        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(
            flow.authorized_scopes().await.unwrap(),
            vec!["challenge.scope", "offline_access", "extra.scope"]
        );
        assert_eq!(flow.get_token().await.unwrap(), "access-token-1");
    }

    #[tokio::test]
    async fn persisted_pkce_state_can_complete_after_flow_rebuild() {
        let http = MockOAuthHttp::new(RegistrationMode::Dynamic);
        let state_store = MemoryOAuthAuthorizationStateStore::new();
        let token_store = MemoryOAuthTokenStore::new();
        let registrations = super::super::oauth_authcode::MemoryOAuthClientRegistrationStore::new();
        let first = flow_builder(
            http.clone(),
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .state_store(state_store.clone())
        .token_store(token_store.clone())
        .registration_store(registrations.clone())
        .build()
        .unwrap();
        let OAuthAuthorizationStart::Pending(pending) = first.begin(["prm.scope"]).await.unwrap()
        else {
            panic!("expected pending flow")
        };
        let request = pending.request().clone();
        let authorization_url = reqwest::Url::parse(&request.authorization_url).unwrap();
        let state = authorization_url
            .query_pairs()
            .find(|(name, _)| name == "state")
            .unwrap()
            .1
            .into_owned();
        drop(pending);

        let second = flow_builder(
            http,
            dynamic_options(),
            AutomaticAuthorizationHandler::default(),
        )
        .state_store(state_store)
        .token_store(token_store)
        .registration_store(registrations)
        .build()
        .unwrap();
        let mut callback = reqwest::Url::parse(&request.redirect_uri).unwrap();
        callback
            .query_pairs_mut()
            .append_pair("code", "persisted-code")
            .append_pair("state", &state)
            .append_pair("iss", &request.issuer);
        second
            .complete_callback_url(callback.as_str())
            .await
            .unwrap();
        assert_eq!(second.get_token().await.unwrap(), "access-token-0");
    }

    #[test]
    fn token_without_lifetime_remains_valid_until_the_server_rejects_it() {
        let token = token_from_response(
            MockOAuthHttp::response(
                200,
                serde_json::json!({
                    "access_token": "access-token",
                    "token_type": "Bearer"
                }),
            ),
            &["read".to_string()],
            None,
        )
        .unwrap();

        assert_eq!(token.expires_at, u64::MAX);
        assert!(token_is_valid(&token, Duration::from_secs(30)));
    }

    #[test]
    fn token_response_rejects_non_bearer_token_types() {
        let error = token_from_response(
            MockOAuthHttp::response(
                200,
                serde_json::json!({
                    "access_token": "access-token",
                    "token_type": "DPoP"
                }),
            ),
            &[],
            None,
        )
        .unwrap_err();

        assert!(error.to_string().contains("unsupported token type `DPoP`"));
    }

    #[test]
    fn scope_selection_falls_back_from_challenge_to_resource_metadata() {
        let metadata: OAuthAuthorizationServerMetadata =
            serde_json::from_value(serde_json::json!({
                "issuer": "https://auth.example.com/issuer",
                "authorization_endpoint": "https://auth.example.com/authorize",
                "token_endpoint": "https://auth.example.com/token",
                "grant_types_supported": ["authorization_code"],
                "scopes_supported": ["as.scope"]
            }))
            .unwrap();
        let protected_resource: OAuthProtectedResourceMetadata =
            serde_json::from_value(serde_json::json!({
                "resource": "https://mcp.example.com/mcp",
                "authorization_servers": ["https://auth.example.com/issuer"],
                "scopes_supported": ["prm.scope"]
            }))
            .unwrap();
        let mut discovery = FlowDiscovery {
            resource: protected_resource.resource.clone(),
            protected_resource,
            authorization_servers: vec![metadata.clone()],
            challenge: None,
        };

        assert_eq!(select_scopes(&[], &discovery, &metadata), ["prm.scope"]);
        // With nothing from the challenge or the resource, the AS's own
        // `scopes_supported` is not used; `scope` is omitted.
        discovery.protected_resource.scopes_supported.clear();
        assert!(select_scopes(&[], &discovery, &metadata).is_empty());
        assert_eq!(
            select_scopes(&["explicit.scope".to_string()], &discovery, &metadata),
            ["explicit.scope"]
        );
        discovery.protected_resource.scopes_supported = vec!["prm.scope".to_string()];
        discovery.challenge = Some(OAuthBearerChallenge {
            error: None,
            scopes: vec!["challenge.scope".to_string()],
            resource_metadata: None,
            error_description: None,
        });
        assert_eq!(
            select_scopes(&[], &discovery, &metadata),
            ["challenge.scope"]
        );
        assert_eq!(
            select_scopes(&["explicit.scope".to_string()], &discovery, &metadata),
            ["explicit.scope"]
        );
    }

    #[test]
    fn resource_identifier_may_be_a_canonical_parent_on_the_same_origin() {
        assert!(
            validate_resource_identifier("https://mcp.example.com/mcp", "https://mcp.example.com")
                .is_ok()
        );
        assert!(
            validate_resource_identifier(
                "https://mcp.example.com/tenant/mcp",
                "https://mcp.example.com/tenant"
            )
            .is_ok()
        );
        assert!(
            validate_resource_identifier(
                "https://mcp.example.com/tenant-evil/mcp",
                "https://mcp.example.com/tenant"
            )
            .is_err()
        );
        assert!(
            validate_resource_identifier(
                "https://other.example.com/mcp",
                "https://mcp.example.com"
            )
            .is_err()
        );
    }

    /// Minimal [`OAuthHttpClient`] used only to test PRM candidate handling:
    /// maps a request path to a canned `(status, body)` response. A path
    /// with no entry panics, so an unexpected candidate request is caught
    /// immediately rather than hanging or silently mismatching.
    struct MockDiscoveryHttp {
        responses: HashMap<&'static str, (u16, String)>,
    }

    #[async_trait]
    impl OAuthHttpClient for MockDiscoveryHttp {
        async fn execute(
            &self,
            request: OAuthHttpRequest,
        ) -> Result<OAuthHttpResponse, OAuthClientError> {
            let path = reqwest::Url::parse(&request.url)
                .unwrap()
                .path()
                .to_string();
            let (status, body) = self
                .responses
                .get(path.as_str())
                .unwrap_or_else(|| panic!("unexpected request path: {path}"));
            Ok(OAuthHttpResponse {
                status: *status,
                headers: Vec::new(),
                body: body.clone().into_bytes(),
            })
        }
    }

    fn no_challenge_metadata() -> Option<OAuthBearerChallenge> {
        Some(OAuthBearerChallenge {
            error: None,
            scopes: Vec::new(),
            resource_metadata: None,
            error_description: None,
        })
    }

    fn authorization_server_metadata_json(issuer: &str) -> String {
        serde_json::json!({
            "issuer": issuer,
            "authorization_endpoint": format!("{issuer}/authorize"),
            "token_endpoint": format!("{issuer}/token"),
            "code_challenge_methods_supported": ["S256"],
            "token_endpoint_auth_methods_supported": ["none"]
        })
        .to_string()
    }

    #[tokio::test]
    async fn discover_with_http_skips_a_mismatched_candidate_and_uses_the_next_one() {
        let http = MockDiscoveryHttp {
            responses: HashMap::from([
                (
                    "/.well-known/oauth-protected-resource/mcp",
                    (
                        200,
                        serde_json::json!({
                            "resource": "https://mcp.example.com/catch-all",
                            "authorization_servers": ["https://auth.example.com/issuer"]
                        })
                        .to_string(),
                    ),
                ),
                (
                    "/.well-known/oauth-protected-resource",
                    (
                        200,
                        serde_json::json!({
                            "resource": "https://mcp.example.com/mcp",
                            "authorization_servers": ["https://auth.example.com/issuer"]
                        })
                        .to_string(),
                    ),
                ),
                (
                    "/.well-known/oauth-authorization-server/issuer",
                    (
                        200,
                        authorization_server_metadata_json("https://auth.example.com/issuer"),
                    ),
                ),
            ]),
        };

        let discovery = discover_with_http(
            "https://mcp.example.com/mcp",
            no_challenge_metadata(),
            &http,
        )
        .await
        .unwrap();

        assert_eq!(discovery.resource, "https://mcp.example.com/mcp");
    }

    #[tokio::test]
    async fn discover_with_http_falls_back_to_well_known_when_the_challenge_url_404s() {
        let http = MockDiscoveryHttp {
            responses: HashMap::from([
                ("/challenge-metadata", (404, String::new())),
                (
                    "/.well-known/oauth-protected-resource/mcp",
                    (
                        200,
                        serde_json::json!({
                            "resource": "https://mcp.example.com/mcp",
                            "authorization_servers": ["https://auth.example.com/issuer"]
                        })
                        .to_string(),
                    ),
                ),
                (
                    "/.well-known/oauth-authorization-server/issuer",
                    (
                        200,
                        authorization_server_metadata_json("https://auth.example.com/issuer"),
                    ),
                ),
            ]),
        };

        let challenge = Some(OAuthBearerChallenge {
            error: None,
            scopes: Vec::new(),
            resource_metadata: Some("https://mcp.example.com/challenge-metadata".to_string()),
            error_description: None,
        });

        let discovery = discover_with_http("https://mcp.example.com/mcp", challenge, &http)
            .await
            .unwrap();

        assert_eq!(discovery.resource, "https://mcp.example.com/mcp");
    }
}