//! Backend management

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

use dashmap::DashMap;
use parking_lot::RwLock;
use reqwest::Client;
use serde_json::Value;
use tokio::sync::{Mutex, Semaphore, watch};
use tracing::{debug, info, warn};

use crate::config::{BackendConfig, RuntimeConfig, TransportConfig};
use crate::failsafe::{Failsafe, with_retry};
use crate::oauth::{OAuthClient, OAuthClientConfig, TokenStorage};
use crate::protocol::{
    JsonRpcResponse, Prompt, PromptsListResult, Resource, ResourceTemplate, ResourcesListResult,
    ResourcesTemplatesListResult, Tool, ToolAnnotations, ToolsListResult,
};
use crate::runtime::{
    RuntimeDenyReason, RuntimeLaunchCommand, RuntimeLaunchMode, RuntimeLicenseTier, RuntimePlan,
    RuntimeProviderKind,
};
use crate::transport::{HttpTransport, StdioTransport, Transport};
use crate::{Error, Result};

struct CachedMetadata<T> {
    state: RwLock<CachedMetadataState<T>>,
}

struct CachedMetadataState<T> {
    value: Option<Arc<T>>,
    cached_at: Option<Instant>,
    in_flight: Option<watch::Sender<()>>,
}

impl<T> Default for CachedMetadataState<T> {
    fn default() -> Self {
        Self {
            value: None,
            cached_at: None,
            in_flight: None,
        }
    }
}

enum CacheFetchState<'a, T> {
    Cached(Arc<T>),
    Wait(watch::Receiver<()>),
    Fetch(FetchPermit<'a, T>),
}

struct FetchPermit<'a, T> {
    cache: &'a CachedMetadata<T>,
    sender: watch::Sender<()>,
}

impl<T> Drop for FetchPermit<'_, T> {
    fn drop(&mut self) {
        self.cache.state.write().in_flight = None;
        let _ = self.sender.send(());
    }
}

impl<T> CachedMetadata<T> {
    fn new() -> Self {
        Self {
            state: RwLock::new(CachedMetadataState::default()),
        }
    }

    fn with_cached<R>(&self, map: impl FnOnce(Option<&Arc<T>>) -> R) -> R {
        let state = self.state.read();
        map(state.value.as_ref())
    }

    fn is_fresh(&self, ttl: Duration) -> bool {
        let state = self.state.read();
        matches!(
            (&state.value, state.cached_at),
            (Some(_), Some(cached_at)) if cached_at.elapsed() < ttl
        )
    }

    fn snapshot_shared(&self) -> Option<Arc<T>> {
        let state = self.state.read();
        state.value.clone()
    }

    fn store_shared(&self, value: Arc<T>) {
        let mut state = self.state.write();
        state.value = Some(value);
        state.cached_at = Some(Instant::now());
    }

    fn acquire(&self, ttl: Duration) -> CacheFetchState<'_, T> {
        {
            let state = self.state.read();
            if let Some(value) = Self::fresh_value(&state, ttl) {
                return CacheFetchState::Cached(value);
            }
            if let Some(sender) = state.in_flight.as_ref() {
                return CacheFetchState::Wait(sender.subscribe());
            }
        }

        let mut state = self.state.write();
        if let Some(value) = Self::fresh_value(&state, ttl) {
            return CacheFetchState::Cached(value);
        }
        if let Some(sender) = state.in_flight.as_ref() {
            return CacheFetchState::Wait(sender.subscribe());
        }

        let (sender, _receiver) = watch::channel(());
        state.in_flight = Some(sender.clone());
        CacheFetchState::Fetch(FetchPermit {
            cache: self,
            sender,
        })
    }

    async fn get_or_fetch_shared<F, Fut>(&self, ttl: Duration, fetch: F) -> Result<Arc<T>>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        loop {
            match self.acquire(ttl) {
                CacheFetchState::Cached(value) => return Ok(value),
                CacheFetchState::Wait(mut receiver) => {
                    let _ = receiver.changed().await;
                }
                CacheFetchState::Fetch(permit) => {
                    let result = fetch().await.map(Arc::new);
                    if let Ok(value) = &result {
                        self.store_shared(Arc::clone(value));
                    }
                    drop(permit);
                    return result;
                }
            }
        }
    }

    fn fresh_value(state: &CachedMetadataState<T>, ttl: Duration) -> Option<Arc<T>> {
        if let (Some(value), Some(cached_at)) = (&state.value, state.cached_at)
            && cached_at.elapsed() < ttl
        {
            return Some(Arc::clone(value));
        }

        None
    }
}

/// Compile the runtime profile selected by a backend into a live-start plan.
#[must_use]
pub fn runtime_plan_for_backend(
    name: &str,
    config: &BackendConfig,
    runtime_config: &RuntimeConfig,
) -> Option<RuntimePlan> {
    let profile_name = config.runtime_profile.as_deref()?;
    let executable_hint = stdio_executable_hint(&config.transport);
    runtime_config.plan_backend_profile(profile_name, name, executable_hint.as_deref())
}

fn stdio_executable_hint(transport: &TransportConfig) -> Option<String> {
    let TransportConfig::Stdio { command, .. } = transport else {
        return None;
    };
    shlex::split(command)?.into_iter().next()
}

/// MCP Backend - manages connection to a single MCP server
pub struct Backend {
    /// Backend name
    pub name: String,
    /// Configuration
    config: BackendConfig,
    /// Runtime plan compiled from the backend's configured runtime profile.
    runtime_plan: Option<RuntimePlan>,
    /// Transport
    transport: RwLock<Option<Arc<dyn Transport>>>,
    /// Serializes backend startup so concurrent warm-start/client requests do
    /// not spawn duplicate stdio processes for the same backend.
    start_lock: Mutex<()>,
    /// Failsafe mechanisms
    failsafe: Failsafe,
    /// Cached tools
    tools_cache: CachedMetadata<Vec<Tool>>,
    /// Cached resources
    resources_cache: CachedMetadata<Vec<Resource>>,
    /// Cached resource templates
    resource_templates_cache: CachedMetadata<Vec<ResourceTemplate>>,
    /// Cached prompts
    prompts_cache: CachedMetadata<Vec<Prompt>>,
    /// Cache TTL
    cache_ttl: Duration,
    /// Last used timestamp
    last_used: AtomicU64,
    /// Concurrency limiter
    semaphore: Semaphore,
    /// Request counter
    request_count: AtomicU64,
}

impl Backend {
    /// Create a new backend
    #[must_use]
    pub fn new(
        name: &str,
        config: BackendConfig,
        failsafe_config: &crate::config::FailsafeConfig,
        cache_ttl: Duration,
    ) -> Self {
        Self::new_with_runtime_plan(name, config, failsafe_config, cache_ttl, None)
    }

    /// Create a new backend with an optional precompiled runtime plan.
    #[must_use]
    pub fn new_with_runtime_plan(
        name: &str,
        config: BackendConfig,
        failsafe_config: &crate::config::FailsafeConfig,
        cache_ttl: Duration,
        runtime_plan: Option<RuntimePlan>,
    ) -> Self {
        Self {
            name: name.to_string(),
            config,
            runtime_plan,
            transport: RwLock::new(None),
            start_lock: Mutex::new(()),
            failsafe: Failsafe::new(name, failsafe_config),
            tools_cache: CachedMetadata::new(),
            resources_cache: CachedMetadata::new(),
            resource_templates_cache: CachedMetadata::new(),
            prompts_cache: CachedMetadata::new(),
            cache_ttl,
            last_used: AtomicU64::new(0),
            semaphore: Semaphore::new(100), // Max concurrent requests
            request_count: AtomicU64::new(0),
        }
    }

    /// Ensure backend is started
    ///
    /// # Errors
    ///
    /// Returns an error if the transport fails to start.
    pub async fn ensure_started(&self) -> Result<()> {
        // Update last used
        self.last_used.store(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs(),
            Ordering::Relaxed,
        );

        // Check if already connected
        {
            let transport = self.transport.read();
            if transport.as_ref().is_some_and(|t| t.is_connected()) {
                return Ok(());
            }
        }

        let _start_guard = self.start_lock.lock().await;

        {
            let transport = self.transport.read();
            if transport.as_ref().is_some_and(|t| t.is_connected()) {
                return Ok(());
            }
        }

        // Start transport
        self.start().await
    }

    /// Start the backend
    ///
    /// # Errors
    ///
    /// Returns an error if the transport fails to connect or initialize.
    pub async fn start(&self) -> Result<()> {
        info!(backend = %self.name, "Starting backend");

        let transport: Arc<dyn Transport> = match &self.config.transport {
            TransportConfig::Stdio {
                command,
                cwd,
                protocol_version,
            } => {
                let launch = self.resolve_stdio_runtime_launch(command)?;
                let transport = StdioTransport::new(
                    &launch.command,
                    launch.env,
                    cwd.clone(),
                    self.config.timeout,
                    protocol_version.clone(),
                );
                transport.start().await?;
                transport
            }
            TransportConfig::Http {
                http_url,
                streamable_http,
                protocol_version,
            } => {
                // Create OAuth client if configured
                let oauth_client = self.create_oauth_client(http_url)?;

                let transport = HttpTransport::new_with_oauth(
                    http_url,
                    self.config.headers.clone(),
                    self.config.timeout,
                    *streamable_http,
                    oauth_client,
                    protocol_version.clone(),
                )?;
                transport.initialize().await?;
                transport
            }
            #[cfg(feature = "a2a")]
            TransportConfig::A2a { a2a_url, .. } => {
                // A2A backends are managed by A2aProvider, not the legacy
                // Backend/Transport stack.  Reaching this branch means an A2A
                // backend was incorrectly started through the legacy path.
                return Err(crate::Error::Config(format!(
                    "A2A backend '{name}' (url: {a2a_url}) must be started via A2aProvider, \
                     not the legacy Backend::start() path",
                    name = self.name,
                )));
            }
        };

        *self.transport.write() = Some(transport);

        // Note: Tools are fetched lazily on first get_tools() call
        // We can't pre-cache here because get_tools() -> ensure_started() -> start()
        // would create infinite async recursion

        Ok(())
    }

    /// Create OAuth client if OAuth is configured for this backend
    fn create_oauth_client(&self, resource_url: &str) -> Result<Option<OAuthClient>> {
        let oauth_config = match &self.config.oauth {
            Some(cfg) if cfg.enabled => cfg,
            _ => return Ok(None),
        };

        // F3 sink-side guard. Config::validate() rejects this pairing at load,
        // but programmatic `Backend::new*()` and hot-reload `apply_patch()` build
        // backends from a raw BackendConfig without revalidating. Enforce again
        // here — the last chokepoint before an OAuth client is created — so an
        // enabled backend OAuth client is never spun up alongside
        // identity_propagation. The backend OAuth persists a gateway-held token
        // during initialize(), authenticating the transport session as the
        // gateway before any per-request per-user override, silently defeating
        // per-user propagation. Fail closed at the sink.
        if self.config.identity_propagation.is_some() {
            return Err(Error::ConfigValidation(format!(
                "backend '{}' cannot combine identity_propagation with its own enabled oauth \
                 client: the backend oauth persists a gateway-held token during initialize(), \
                 authenticating the transport session as the gateway before the per-request \
                 credential override — silently defeating per-user propagation (F3).",
                self.name
            )));
        }

        info!(backend = %self.name, "Initializing OAuth client");

        // Create HTTP client for OAuth requests
        let http_client = Client::builder()
            .timeout(Duration::from_secs(30))
            .build()
            .map_err(|e| Error::OAuth(format!("Failed to create OAuth HTTP client: {e}")))?;

        // Get or create token storage
        let storage = Arc::new(
            TokenStorage::default_location()
                .map_err(|e| Error::OAuth(format!("Failed to create token storage: {e}")))?,
        );

        // Create OAuth client
        let oauth = OAuthClient::new(
            http_client,
            self.name.clone(),
            resource_url.to_string(),
            oauth_config.scopes.clone(),
            storage,
            OAuthClientConfig {
                client_id: oauth_config.client_id.clone(),
                client_secret: oauth_config.client_secret.clone(),
                callback_host: oauth_config.callback_host.clone(),
                callback_port: oauth_config.callback_port,
                callback_path: oauth_config.callback_path.clone(),
                token_refresh_buffer_secs: oauth_config.token_refresh_buffer_secs,
            },
        );

        Ok(Some(oauth))
    }

    fn resolve_stdio_runtime_launch(
        &self,
        configured_command: &str,
    ) -> Result<ResolvedStdioLaunch> {
        let Some(plan) = self.runtime_plan.as_ref() else {
            return Ok(ResolvedStdioLaunch {
                command: configured_command.to_string(),
                env: self.config.env.clone(),
            });
        };
        self.enforce_stdio_runtime_plan(plan)?;

        match plan.provider {
            RuntimeProviderKind::LocalProcess => {
                info!(
                    backend = %self.name,
                    provider = ?plan.provider,
                    policy_id = %plan.policy.id,
                    "RuntimeProvider profile accepted before stdio backend start"
                );
                Ok(ResolvedStdioLaunch {
                    command: configured_command.to_string(),
                    env: self.config.env.clone(),
                })
            }
            RuntimeProviderKind::Docker | RuntimeProviderKind::Podman => {
                let command = container_stdio_bridge_command(plan)?;
                info!(
                    backend = %self.name,
                    provider = ?plan.provider,
                    policy_id = %plan.policy.id,
                    "RuntimeProvider container stdio bridge accepted before backend start"
                );
                Ok(ResolvedStdioLaunch {
                    command,
                    env: filter_runtime_env(&self.config.env, &plan.policy.env.allowed_keys),
                })
            }
            RuntimeProviderKind::Systemd
            | RuntimeProviderKind::Launchd
            | RuntimeProviderKind::Kubernetes => Err(Error::Config(format!(
                "backend '{}' runtime profile selected {:?}, but live stdio backend lifecycle currently supports local_process plus docker/podman stdio bridge",
                self.name, plan.provider
            ))),
        }
    }

    fn enforce_stdio_runtime_plan(&self, plan: &RuntimePlan) -> Result<()> {
        if plan.is_denied() {
            let reasons = plan
                .denied
                .iter()
                .map(|denial| format!("{:?}", denial.reason))
                .collect::<Vec<_>>()
                .join(", ");
            return Err(Error::Config(format!(
                "backend '{}' runtime profile '{}' denied by policy: {reasons}",
                self.name, plan.policy.id
            )));
        }
        if plan.requires_confirmation() {
            let confirmations = plan
                .confirmations
                .iter()
                .map(|confirmation| confirmation.id.as_str())
                .collect::<Vec<_>>()
                .join(", ");
            return Err(Error::Config(format!(
                "backend '{}' runtime profile '{}' requires confirmations before live start: {confirmations}",
                self.name, plan.policy.id
            )));
        }
        Ok(())
    }

    /// Stop the backend
    ///
    /// # Errors
    ///
    /// Returns an error if the transport fails to close cleanly.
    pub async fn stop(&self) -> Result<()> {
        info!(backend = %self.name, "Stopping backend");

        let transport = self.transport.write().take();
        if let Some(t) = transport {
            t.close().await?;
        }

        Ok(())
    }

    /// Check if backend is running
    pub fn is_running(&self) -> bool {
        self.transport
            .read()
            .as_ref()
            .is_some_and(|t| t.is_connected())
    }

    /// Get cached tools (or fetch if needed)
    ///
    /// Check if this backend has cached tools (non-blocking).
    ///
    /// Returns `true` if tools are cached and the cache hasn't expired.
    /// Used by `search_tools` to skip unstarted backends.
    #[must_use]
    pub fn has_cached_tools(&self) -> bool {
        self.tools_cache.is_fresh(self.cache_ttl)
    }

    /// Return the number of tools in the cache (non-blocking, no network I/O).
    ///
    /// Returns `0` when the cache is empty or has never been populated.
    /// This is intentionally best-effort: it reads whatever is in the cache
    /// without triggering a refresh, so the count may be stale.
    #[must_use]
    pub fn cached_tools_count(&self) -> usize {
        self.tools_cache
            .with_cached(|tools| tools.map_or(0, |tools| tools.len()))
    }

    /// Return the names of all cached tools (non-blocking, no network I/O).
    ///
    /// Returns an empty `Vec` when the cache is empty or has never been populated.
    /// Intended for producing "did you mean?" suggestions on unknown tool names.
    #[must_use]
    pub fn get_cached_tool_names(&self) -> Vec<String> {
        self.tools_cache.with_cached(|tools| {
            tools
                .map(|tools| tools.iter().map(|t| t.name.clone()).collect())
                .unwrap_or_default()
        })
    }

    /// Return a single tool by exact name from the cache (non-blocking, no network I/O).
    ///
    /// Returns `None` when the cache is empty, has never been populated, or does
    /// not contain a tool with the given name.  Intended for resolving surfaced
    /// tool schemas at `tools/list` time.
    #[must_use]
    pub fn get_cached_tool(&self, name: &str) -> Option<Tool> {
        self.tools_cache.with_cached(|tools| {
            tools.and_then(|tools| tools.iter().find(|t| t.name == name).cloned())
        })
    }

    /// Return a snapshot of all cached tools (non-blocking, no network I/O).
    ///
    /// Returns an empty shared vector when the cache is empty or has never been
    /// populated. Used by the `spec-preview` filtered `tools/list`
    /// implementation to avoid cloning the full tool list on every cache hit.
    #[must_use]
    pub fn get_cached_tools_snapshot(&self) -> Arc<Vec<Tool>> {
        self.tools_cache
            .snapshot_shared()
            .unwrap_or_else(|| Arc::new(Vec::new()))
    }

    async fn get_cached_list_shared<T, F>(
        &self,
        cache: &CachedMetadata<Vec<T>>,
        method: &str,
        kind: &'static str,
        parse: F,
    ) -> Result<Arc<Vec<T>>>
    where
        F: Fn(Value) -> Result<Vec<T>>,
    {
        cache
            .get_or_fetch_shared(self.cache_ttl, || async {
                self.ensure_started().await?;

                let response = self.request_internal(method, None).await?;
                if let Some(error) = response.error {
                    return Err(Error::json_rpc(error.code, error.message));
                }
                let items = if let Some(result) = response.result {
                    parse(result)?
                } else {
                    Vec::new()
                };

                debug!(backend = %self.name, kind, count = items.len(), "Backend metadata cached");

                Ok(items)
            })
            .await
    }

    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the tools request fails.
    pub async fn get_tools_shared(&self) -> Result<Arc<Vec<Tool>>> {
        self.get_cached_list_shared(&self.tools_cache, "tools/list", "tools", |result| {
            let mut tools = serde_json::from_value::<ToolsListResult>(result)?.tools;
            normalize_tool_annotations(&self.name, &mut tools);
            Ok(tools)
        })
        .await
    }

    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the tools request fails.
    pub async fn get_tools(&self) -> Result<Vec<Tool>> {
        self.get_tools_shared()
            .await
            .map(|tools| tools.as_ref().clone())
    }

    /// Get cached resources (or fetch if needed) without cloning the cached list.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the resources request fails.
    pub async fn get_resources_shared(&self) -> Result<Arc<Vec<Resource>>> {
        self.get_cached_list_shared(
            &self.resources_cache,
            "resources/list",
            "resources",
            |result| Ok(serde_json::from_value::<ResourcesListResult>(result)?.resources),
        )
        .await
    }

    /// Get cached resources (or fetch if needed)
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the resources request fails.
    pub async fn get_resources(&self) -> Result<Vec<Resource>> {
        self.get_resources_shared()
            .await
            .map(|resources| resources.as_ref().clone())
    }

    /// Get cached resource templates (or fetch if needed) without cloning the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the templates request fails.
    pub async fn get_resource_templates_shared(&self) -> Result<Arc<Vec<ResourceTemplate>>> {
        self.get_cached_list_shared(
            &self.resource_templates_cache,
            "resources/templates/list",
            "resource_templates",
            |result| {
                Ok(
                    serde_json::from_value::<ResourcesTemplatesListResult>(result)?
                        .resource_templates,
                )
            },
        )
        .await
    }

    /// Get cached resource templates (or fetch if needed)
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the templates request fails.
    pub async fn get_resource_templates(&self) -> Result<Vec<ResourceTemplate>> {
        self.get_resource_templates_shared()
            .await
            .map(|templates| templates.as_ref().clone())
    }

    /// Get cached prompts (or fetch if needed) without cloning the cached list.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the prompts request fails.
    pub async fn get_prompts_shared(&self) -> Result<Arc<Vec<Prompt>>> {
        self.get_cached_list_shared(&self.prompts_cache, "prompts/list", "prompts", |result| {
            Ok(serde_json::from_value::<PromptsListResult>(result)?.prompts)
        })
        .await
    }

    /// Get cached prompts (or fetch if needed)
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot start or the prompts request fails.
    pub async fn get_prompts(&self) -> Result<Vec<Prompt>> {
        self.get_prompts_shared()
            .await
            .map(|prompts| prompts.as_ref().clone())
    }

    /// Internal request without `ensure_started` (to avoid recursion)
    async fn request_internal(
        &self,
        method: &str,
        params: Option<Value>,
    ) -> Result<JsonRpcResponse> {
        // Get transport
        let transport = {
            let t = self.transport.read();
            t.clone()
        };

        let transport = transport.ok_or_else(|| Error::BackendUnavailable(self.name.clone()))?;

        transport.request(method, params).await
    }

    /// Internal notification send without `ensure_started` (to avoid recursion)
    async fn notify_internal(&self, method: &str, params: Option<Value>) -> Result<()> {
        let transport = {
            let t = self.transport.read();
            t.clone()
        };

        let transport = transport.ok_or_else(|| Error::BackendUnavailable(self.name.clone()))?;

        transport.notify(method, params).await
    }

    /// Send a request to the backend
    ///
    /// # Errors
    ///
    /// Returns an error if the backend is unavailable, the concurrency limit
    /// is reached, or the request itself fails after retries.
    #[tracing::instrument(
        skip(self, params),
        fields(
            backend = %self.name,
            method = %method,
            request_id = %uuid::Uuid::new_v4()
        )
    )]
    pub async fn request(&self, method: &str, params: Option<Value>) -> Result<JsonRpcResponse> {
        self.request_with_headers(method, params, &[]).await
    }

    /// This backend's end-user identity-propagation config, if configured
    /// (MIK-6704 / ADR-007). `None` → static-credential behavior unchanged.
    #[must_use]
    pub fn identity_propagation_config(
        &self,
    ) -> Option<&crate::identity_propagation::IdentityPropagationConfig> {
        self.config.identity_propagation.as_ref()
    }

    /// Whether this backend relies on a single gateway-held OAuth token that is
    /// NOT blessed for shared use (ADR-008 INV-2).
    ///
    /// `true` means the gateway stores one token for this backend and would
    /// attach it to any caller's request — unsafe on a multi-user gateway
    /// unless a per-user credential is supplied instead. The dispatch guard
    /// uses this to fail closed. `oauth.shared_account = true` opts out (the
    /// operator has declared the account genuinely shared).
    #[must_use]
    pub fn oauth_requires_per_user_isolation(&self) -> bool {
        self.config
            .oauth
            .as_ref()
            .is_some_and(|o| o.enabled && !o.shared_account)
    }

    /// Send a request, adding per-request outbound headers (e.g. a propagated
    /// end-user identity credential — MIK-6704). The headers are forwarded by
    /// value to the transport's `request_with_headers`, never stored on the
    /// backend, so concurrent per-user requests stay isolated (IDP.3).
    ///
    /// # Errors
    ///
    /// Returns an error if the backend is unavailable, the concurrency limit
    /// is reached, or the request itself fails after retries.
    pub async fn request_with_headers(
        &self,
        method: &str,
        params: Option<Value>,
        extra_headers: &[(String, String)],
    ) -> Result<JsonRpcResponse> {
        let start_time = std::time::Instant::now();

        // Check failsafe
        if !self.failsafe.can_proceed() {
            telemetry_metrics::gauge!(
                "mcp_backend_circuit_state",
                "backend" => self.name.clone()
            )
            .set(0.0_f64);
            tracing::warn!(backend = %self.name, "Request rejected by circuit breaker");
            return Err(Error::CircuitOpen(self.name.clone()));
        }
        telemetry_metrics::gauge!(
            "mcp_backend_circuit_state",
            "backend" => self.name.clone()
        )
        .set(1.0_f64);

        // Acquire semaphore
        let _permit = self.semaphore.acquire().await.map_err(|_| {
            tracing::warn!("Concurrency limit reached");
            Error::BackendUnavailable("Concurrency limit reached".to_string())
        })?;

        self.request_count.fetch_add(1, Ordering::Relaxed);

        // Get transport
        self.ensure_started().await?;

        let transport = {
            let t = self.transport.read();
            t.clone()
        };

        let transport = transport.ok_or_else(|| {
            tracing::error!("Transport not available");
            Error::BackendUnavailable(self.name.clone())
        })?;

        // Execute with retry
        let name = self.name.clone();
        let result = with_retry(&self.failsafe.retry_policy, &name, || {
            let transport = Arc::clone(&transport);
            let method = method.to_string();
            let params = params.clone();
            let extra_headers = extra_headers.to_vec();
            async move {
                transport
                    .request_with_headers(&method, params, &extra_headers)
                    .await
            }
        })
        .await;

        // Calculate latency
        let latency = start_time.elapsed();

        // Record success/failure with metrics
        match &result {
            Ok(_) => {
                tracing::info!(
                    latency_ms = latency.as_millis(),
                    "Request completed successfully"
                );
                self.failsafe.record_success(latency);
                telemetry_metrics::counter!(
                    "mcp_backend_requests_total",
                    "backend" => self.name.clone(),
                    "status" => "ok"
                )
                .increment(1);
            }
            Err(e) => {
                tracing::error!(error = %e, latency_ms = latency.as_millis(), "Request failed");
                self.failsafe.record_failure(&e.to_string(), latency);
                telemetry_metrics::counter!(
                    "mcp_backend_requests_total",
                    "backend" => self.name.clone(),
                    "status" => "error"
                )
                .increment(1);
            }
        }
        telemetry_metrics::histogram!(
            "mcp_backend_request_duration_seconds",
            "backend" => self.name.clone()
        )
        .record(latency.as_secs_f64());

        result
    }

    /// Send a notification to the backend.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend is unavailable, the concurrency limit
    /// is reached, or the notification cannot be sent.
    #[tracing::instrument(
        skip(self, params),
        fields(
            backend = %self.name,
            method = %method,
            request_id = %uuid::Uuid::new_v4()
        )
    )]
    pub async fn notify(&self, method: &str, params: Option<Value>) -> Result<()> {
        let start_time = std::time::Instant::now();

        if !self.failsafe.can_proceed() {
            telemetry_metrics::gauge!(
                "mcp_backend_circuit_state",
                "backend" => self.name.clone()
            )
            .set(0.0_f64);
            tracing::warn!(backend = %self.name, "Notification rejected by circuit breaker");
            return Err(Error::CircuitOpen(self.name.clone()));
        }
        telemetry_metrics::gauge!(
            "mcp_backend_circuit_state",
            "backend" => self.name.clone()
        )
        .set(1.0_f64);

        let _permit = self.semaphore.acquire().await.map_err(|_| {
            tracing::warn!("Concurrency limit reached");
            Error::BackendUnavailable("Concurrency limit reached".to_string())
        })?;

        self.request_count.fetch_add(1, Ordering::Relaxed);

        self.ensure_started().await?;

        let result = self.notify_internal(method, params).await;
        let latency = start_time.elapsed();

        match &result {
            Ok(()) => {
                tracing::info!(
                    latency_ms = latency.as_millis(),
                    "Notification sent successfully"
                );
                self.failsafe.record_success(latency);
                telemetry_metrics::counter!(
                    "mcp_backend_requests_total",
                    "backend" => self.name.clone(),
                    "status" => "ok"
                )
                .increment(1);
            }
            Err(e) => {
                tracing::error!(error = %e, latency_ms = latency.as_millis(), "Notification failed");
                self.failsafe.record_failure(&e.to_string(), latency);
                telemetry_metrics::counter!(
                    "mcp_backend_requests_total",
                    "backend" => self.name.clone(),
                    "status" => "error"
                )
                .increment(1);
            }
        }
        telemetry_metrics::histogram!(
            "mcp_backend_request_duration_seconds",
            "backend" => self.name.clone()
        )
        .record(latency.as_secs_f64());

        result
    }

    #[cfg(test)]
    pub(crate) fn set_transport_for_test(&self, transport: Arc<dyn Transport>) {
        *self.transport.write() = Some(transport);
    }

    /// Test-only: trip this backend's circuit breaker open by recording
    /// `failure_threshold` consecutive failures.
    #[cfg(test)]
    pub(crate) fn trip_circuit_breaker_for_test(&self) {
        let threshold = self.failsafe.circuit_breaker.stats().failure_threshold;
        for _ in 0..threshold {
            self.failsafe
                .circuit_breaker
                .record_failure("test-trip", std::time::Duration::ZERO);
        }
    }

    /// Return `true` if this backend is configured for pass-through mode.
    ///
    /// When `true`, the direct `/mcp/{name}` endpoint skips tool policy
    /// enforcement and input sanitization for `tools/call` requests.
    /// This must only be enabled for fully-trusted internal backends.
    #[must_use]
    pub fn passthrough(&self) -> bool {
        self.config.passthrough
    }

    /// Return the HTTP URL if this backend uses an HTTP-based transport.
    ///
    /// Returns `None` for stdio backends.
    #[must_use]
    pub fn transport_url(&self) -> Option<&str> {
        match &self.config.transport {
            TransportConfig::Http { http_url, .. } => Some(http_url.as_str()),
            TransportConfig::Stdio { .. } => None,
            #[cfg(feature = "a2a")]
            TransportConfig::A2a { a2a_url, .. } => Some(a2a_url.as_str()),
        }
    }

    /// Get backend status
    pub fn status(&self) -> BackendStatus {
        let health = self.failsafe.health_metrics();
        BackendStatus {
            name: self.name.clone(),
            running: self.is_running(),
            transport: self.config.transport.transport_type().to_string(),
            tools_cached: self.cached_tools_count(),
            circuit_state: self.failsafe.circuit_breaker.state().as_str().to_string(),
            request_count: self.request_count.load(Ordering::Relaxed),
            healthy: health.healthy,
            consecutive_failures: health.consecutive_failures,
            latency_p95_ms: health.latency_p95_ms,
            runtime: self.runtime_status(),
        }
    }

    fn runtime_status(&self) -> Option<BackendRuntimeStatus> {
        let plan = self.runtime_plan.as_ref()?;
        let state = if plan.is_denied() {
            BackendRuntimeState::Denied
        } else if plan.requires_confirmation() {
            BackendRuntimeState::ConfirmationRequired
        } else {
            BackendRuntimeState::Ready
        };

        Some(BackendRuntimeStatus {
            profile: self
                .config
                .runtime_profile
                .clone()
                .unwrap_or_else(|| plan.policy.id.clone()),
            provider: plan.provider,
            policy_id: plan.policy.id.clone(),
            license_tier: plan.audit.license_tier,
            state,
            denied_reasons: plan.denied.iter().map(|denial| denial.reason).collect(),
            confirmation_ids: plan
                .confirmations
                .iter()
                .map(|confirmation| confirmation.id.clone())
                .collect(),
            restart_max_attempts: plan.policy.restart.max_restarts,
            restart_backoff_secs: plan.policy.restart.backoff_secs,
            health_check: plan.lifecycle.health_check.clone(),
            restart_command_hint: plan.lifecycle.restart_command_hint.clone(),
            rollback_step: plan.rollback_step.clone(),
        })
    }

    /// Get circuit breaker stats for this backend.
    pub fn circuit_breaker_stats(&self) -> crate::failsafe::CircuitBreakerStats {
        self.failsafe.circuit_breaker.stats()
    }

    /// Force this backend's circuit breaker back to `Closed` (MIK-5983).
    ///
    /// Called by `gateway_revive_server` so the documented manual recovery
    /// path also clears a tripped breaker, not just the kill switch.
    pub fn reset_circuit_breaker(&self) {
        self.failsafe.circuit_breaker.reset();
    }

    /// Whether this backend's circuit breaker is currently tripped
    /// (`Open` or `HalfOpen` — i.e. not `Closed`).
    #[must_use]
    pub fn is_circuit_tripped(&self) -> bool {
        self.failsafe.circuit_breaker.state() != crate::failsafe::CircuitState::Closed
    }

    /// Tear down the current transport (killing any child process) and start a
    /// fresh one.
    ///
    /// Unlike [`ensure_started`](Self::ensure_started), this does **not** trust
    /// `is_connected()` — it always rebuilds. A wedged-but-not-exited child
    /// (responds to `try_wait` as alive yet never answers requests) cannot be
    /// recovered by `ensure_started` alone; this is the escape hatch the health
    /// loop uses when a probe fails.
    ///
    /// # Errors
    ///
    /// Returns an error if the fresh transport fails to start or initialize.
    pub async fn force_restart(&self) -> Result<()> {
        let _guard = self.start_lock.lock().await;
        // Take the transport out and drop the RwLock write guard *before*
        // awaiting close() — a parking_lot guard is not Send across an await.
        let old = self.transport.write().take();
        if let Some(old) = old {
            let _ = old.close().await;
        }
        self.start().await
    }

    /// Active health/recovery probe driven by the background health loop.
    ///
    /// This is the gateway's automatic equivalent of `gateway_revive_server`.
    /// Two properties make it actually recover a wedged backend, which the old
    /// `backend.request("ping")` health check could not:
    ///
    /// 1. **It bypasses the circuit breaker.** A probe routed through
    ///    [`request`](Self::request) short-circuits on `can_proceed()` and
    ///    returns `CircuitOpen` *without touching the backend* — so it could
    ///    never discover that an `Open` backend had recovered. This probe talks
    ///    to the transport directly.
    /// 2. **On success it resets a tripped breaker**; on failure it forces a
    ///    transport rebuild so the next probe targets a fresh child.
    ///
    /// `timeout` bounds the probe so a hung backend cannot stall the loop.
    ///
    /// # Errors
    ///
    /// Returns an error if the backend cannot be started, the probe times out,
    /// or the `ping` call fails. The breaker is left for organic traffic to
    /// trip — this probe never records failures, only recoveries.
    pub async fn health_probe(&self, timeout: Duration) -> Result<()> {
        // `ensure_started` now respawns reliably because `is_connected()` does a
        // real liveness check (Fix C).
        if let Err(e) = self.ensure_started().await {
            let _ = self.force_restart().await;
            return Err(e);
        }

        let transport = { self.transport.read().clone() };
        let Some(transport) = transport else {
            return Err(Error::BackendUnavailable(self.name.clone()));
        };

        match tokio::time::timeout(timeout, transport.request("ping", None)).await {
            Ok(Ok(_)) => {
                if self.is_circuit_tripped() {
                    info!(
                        backend = %self.name,
                        "Health probe succeeded; resetting tripped circuit breaker"
                    );
                    self.failsafe.circuit_breaker.reset();
                }
                Ok(())
            }
            Ok(Err(e)) => {
                warn!(backend = %self.name, error = %e, "Health probe failed; rebuilding transport");
                let _ = self.force_restart().await;
                Err(e)
            }
            Err(_elapsed) => {
                warn!(
                    backend = %self.name,
                    timeout_ms = timeout.as_millis(),
                    "Health probe timed out; rebuilding transport"
                );
                let _ = self.force_restart().await;
                Err(Error::BackendTimeout(self.name.clone()))
            }
        }
    }

    /// Get health metrics for this backend.
    pub fn health_metrics(&self) -> crate::failsafe::HealthMetrics {
        self.failsafe.health_metrics()
    }
}

struct ResolvedStdioLaunch {
    command: String,
    env: HashMap<String, String>,
}

fn container_stdio_bridge_command(plan: &RuntimePlan) -> Result<String> {
    let command = plan.launch_command.as_ref().ok_or_else(|| {
        Error::Config(format!(
            "runtime provider {:?} has no structured launch command for stdio bridge",
            plan.provider
        ))
    })?;
    if command.args.first().map(String::as_str) != Some("run") {
        return Err(Error::Config(format!(
            "runtime provider {:?} launch command is not a container run command",
            plan.provider
        )));
    }

    let mut args = vec![
        "run".to_string(),
        "--interactive".to_string(),
        "--rm".to_string(),
    ];
    let mut skip_restart_value = false;
    for arg in command.args.iter().skip(1) {
        if skip_restart_value {
            skip_restart_value = false;
            continue;
        }
        match arg.as_str() {
            "--detach" | "-d" | "--interactive" | "-i" | "--rm" => {}
            "--restart" => skip_restart_value = true,
            value if value.starts_with("--restart=") => {}
            _ => args.push(arg.clone()),
        }
    }

    Ok(RuntimeLaunchCommand {
        program: command.program.clone(),
        args,
        mode: RuntimeLaunchMode::RunToCompletion,
    }
    .display_command())
}

fn filter_runtime_env(
    env: &HashMap<String, String>,
    allowed_keys: &[String],
) -> HashMap<String, String> {
    allowed_keys
        .iter()
        .filter_map(|key| env.get(key).map(|value| (key.clone(), value.clone())))
        .collect()
}

/// Backend status information
#[derive(Debug, Clone, serde::Serialize)]
pub struct BackendStatus {
    /// Backend name
    pub name: String,
    /// Whether backend is running
    pub running: bool,
    /// Transport type
    pub transport: String,
    /// Number of cached tools
    pub tools_cached: usize,
    /// Circuit breaker state
    pub circuit_state: String,
    /// Total request count
    pub request_count: u64,
    /// Health-tracker liveness (flips false after consecutive failures, e.g.
    /// timeouts under load, *before* the circuit breaker trips Open). `/health`
    /// must consider this so it does not report healthy while a backend is
    /// silently timing out (see issue #5080 / MIK-5080).
    pub healthy: bool,
    /// Consecutive failures recorded by the health tracker.
    pub consecutive_failures: u64,
    /// 95th percentile latency in milliseconds, if any samples exist.
    pub latency_p95_ms: Option<u64>,
    /// Runtime profile lifecycle state for admin/operator surfaces.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub runtime: Option<BackendRuntimeStatus>,
}

/// Runtime profile status information exposed through backend status.
#[derive(Debug, Clone, serde::Serialize, PartialEq, Eq)]
pub struct BackendRuntimeStatus {
    /// Runtime profile selected by this backend.
    pub profile: String,
    /// Provider selected by the compiled runtime plan.
    pub provider: RuntimeProviderKind,
    /// Policy id used for audit correlation.
    pub policy_id: String,
    /// License tier that owns this runtime provider capability.
    pub license_tier: RuntimeLicenseTier,
    /// Whether the runtime plan is ready, denied, or waiting for approval.
    pub state: BackendRuntimeState,
    /// Fail-closed denial reasons, when any.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub denied_reasons: Vec<RuntimeDenyReason>,
    /// Confirmation ids required before live start or provider apply.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub confirmation_ids: Vec<String>,
    /// Maximum restart attempts from the compiled policy.
    pub restart_max_attempts: u32,
    /// Restart backoff from the compiled policy.
    pub restart_backoff_secs: u64,
    /// Provider-specific health check instruction or command.
    pub health_check: String,
    /// Provider-specific restart command hint, when available.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_command_hint: Option<String>,
    /// Rollback instruction for this runtime plan.
    pub rollback_step: String,
}

/// Compiled runtime plan state for a backend.
#[derive(Debug, Clone, Copy, serde::Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum BackendRuntimeState {
    /// Policy passed without pending human gates.
    Ready,
    /// Policy requires explicit human approval before execution.
    ConfirmationRequired,
    /// Policy denied execution and must fail closed.
    Denied,
}

/// Backend registry - manages all backends
pub struct BackendRegistry {
    /// Backends by name
    backends: DashMap<String, Arc<Backend>>,
}

impl BackendRegistry {
    /// Create a new registry
    #[must_use]
    pub fn new() -> Self {
        Self {
            backends: DashMap::new(),
        }
    }

    /// Register a backend
    pub fn register(&self, backend: Arc<Backend>) {
        self.backends.insert(backend.name.clone(), backend);
    }

    /// Get a backend by name
    #[must_use]
    pub fn get(&self, name: &str) -> Option<Arc<Backend>> {
        self.backends.get(name).map(|b| Arc::clone(&*b))
    }

    /// Get all backends
    #[must_use]
    pub fn all(&self) -> Vec<Arc<Backend>> {
        self.backends.iter().map(|b| Arc::clone(&*b)).collect()
    }

    /// Get all backend statuses
    #[must_use]
    pub fn statuses(&self) -> HashMap<String, BackendStatus> {
        self.backends
            .iter()
            .map(|b| (b.name.clone(), b.status()))
            .collect()
    }

    /// Remove a backend by name (deregister without stopping).
    ///
    /// If the backend must be stopped before removal, call `backend.stop()`
    /// first.  Returns `true` when the backend was present and removed.
    pub fn remove(&self, name: &str) -> bool {
        self.backends.remove(name).is_some()
    }

    /// Stop all backends
    pub async fn stop_all(&self) {
        for backend in &self.backends {
            if let Err(e) = backend.stop().await {
                warn!(backend = %backend.name, error = %e, "Failed to stop backend");
            }
        }
    }
}

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

pub(crate) fn normalize_tool_annotations(server: &str, tools: &mut [Tool]) {
    for tool in tools {
        let inferred_read_only = infer_read_only_tool(&tool.name);
        let annotations = tool
            .annotations
            .get_or_insert_with(ToolAnnotations::default);
        let read_only = annotations.read_only_hint.unwrap_or(inferred_read_only);
        let destructive = annotations
            .destructive_hint
            .unwrap_or_else(|| infer_destructive_tool(&tool.name, read_only));

        annotations.read_only_hint = Some(read_only);
        annotations.destructive_hint = Some(destructive);
        annotations.idempotent_hint = Some(
            annotations
                .idempotent_hint
                .unwrap_or_else(|| infer_idempotent_tool(&tool.name, read_only, destructive)),
        );
        annotations.open_world_hint = Some(
            annotations
                .open_world_hint
                .unwrap_or_else(|| infer_open_world_tool(server, &tool.name)),
        );
    }
}

fn infer_read_only_tool(name: &str) -> bool {
    let name = name.to_ascii_lowercase();
    let read_prefixes = [
        "analyze",
        "auth_lookup",
        "benchmark",
        "calculate",
        "check",
        "classify",
        "count",
        "describe",
        "detect",
        "estimate",
        "fetch",
        "find",
        "fingerprint",
        "get",
        "health",
        "info",
        "list",
        "lookup",
        "preview",
        "query",
        "read",
        "recall",
        "search",
        "status",
        "suggest",
        "validate",
        "verify",
    ];
    read_prefixes
        .iter()
        .any(|prefix| name == *prefix || name.starts_with(&format!("{prefix}_")))
}

fn infer_destructive_tool(name: &str, read_only: bool) -> bool {
    if read_only {
        return false;
    }

    let name = name.to_ascii_lowercase();
    let destructive_words = [
        "archive", "bash", "clear", "delete", "forget", "kill", "login", "post", "remove", "run",
        "send", "submit", "type", "write",
    ];
    destructive_words.iter().any(|word| name.contains(word))
}

fn infer_idempotent_tool(name: &str, read_only: bool, destructive: bool) -> bool {
    if read_only {
        return true;
    }
    if destructive {
        return false;
    }

    let name = name.to_ascii_lowercase();
    name.starts_with("set_")
        || name.starts_with("clear_")
        || name.starts_with("focus_")
        || name.starts_with("connect")
}

fn infer_open_world_tool(server: &str, name: &str) -> bool {
    let server = server.to_ascii_lowercase();
    let name = name.to_ascii_lowercase();

    if matches!(
        server.as_str(),
        "hebb" | "metacognition" | "pithy" | "cached-grep" | "haiku-file-reader"
    ) {
        return false;
    }

    if name.contains("validate") || name.contains("fingerprint") || name.contains("auth_lookup") {
        return false;
    }

    true
}

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

    use async_trait::async_trait;
    use serde_json::json;
    use tokio::sync::Barrier;
    use tokio::time::sleep;

    use super::*;
    use crate::protocol::{RequestId, ToolsListResult};

    struct MockTransport {
        response: JsonRpcResponse,
        delay: Duration,
        connected: AtomicBool,
        requests: AtomicUsize,
    }

    impl MockTransport {
        fn new(response: JsonRpcResponse, delay: Duration) -> Self {
            Self {
                response,
                delay,
                connected: AtomicBool::new(true),
                requests: AtomicUsize::new(0),
            }
        }
    }

    #[async_trait]
    impl Transport for MockTransport {
        async fn request(&self, method: &str, _params: Option<Value>) -> Result<JsonRpcResponse> {
            assert_eq!(method, "tools/list");
            self.requests.fetch_add(1, Ordering::SeqCst);
            sleep(self.delay).await;
            Ok(self.response.clone())
        }

        async fn notify(&self, _method: &str, _params: Option<Value>) -> Result<()> {
            Ok(())
        }

        fn is_connected(&self) -> bool {
            self.connected.load(Ordering::Relaxed)
        }

        async fn close(&self) -> Result<()> {
            self.connected.store(false, Ordering::Relaxed);
            Ok(())
        }
    }

    // Method-agnostic transport for health-probe / recovery tests: answers any
    // request with success unless `fail` is set, with a settable `connected`
    // flag. Distinct from MockTransport, which hard-asserts "tools/list".
    struct RecoveryMock {
        connected: AtomicBool,
        fail: AtomicBool,
        pings: AtomicUsize,
    }

    impl RecoveryMock {
        fn connected() -> Self {
            Self {
                connected: AtomicBool::new(true),
                fail: AtomicBool::new(false),
                pings: AtomicUsize::new(0),
            }
        }
    }

    #[async_trait]
    impl Transport for RecoveryMock {
        async fn request(&self, _method: &str, _params: Option<Value>) -> Result<JsonRpcResponse> {
            self.pings.fetch_add(1, Ordering::SeqCst);
            if self.fail.load(Ordering::Relaxed) {
                return Err(Error::BackendUnavailable("probe failed".to_string()));
            }
            Ok(JsonRpcResponse::success_serialized(
                RequestId::Number(1),
                json!({}),
            ))
        }

        async fn notify(&self, _method: &str, _params: Option<Value>) -> Result<()> {
            Ok(())
        }

        fn is_connected(&self) -> bool {
            self.connected.load(Ordering::Relaxed)
        }

        async fn close(&self) -> Result<()> {
            self.connected.store(false, Ordering::Relaxed);
            Ok(())
        }
    }

    #[tokio::test]
    async fn is_circuit_tripped_reflects_breaker_state() {
        let backend = Backend::new(
            "test",
            BackendConfig::default(),
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        );
        assert!(!backend.is_circuit_tripped());
        backend.trip_circuit_breaker_for_test();
        assert!(backend.is_circuit_tripped());
        backend.reset_circuit_breaker();
        assert!(!backend.is_circuit_tripped());
    }

    // Headline regression: a successful health probe must auto-reset a tripped
    // breaker. This is the recovery the old health check could never perform,
    // because it pinged through the breaker (which short-circuits when Open).
    #[tokio::test]
    async fn health_probe_resets_tripped_breaker_on_success() {
        let backend = Arc::new(Backend::new(
            "test",
            BackendConfig::default(),
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        ));
        let mock = Arc::new(RecoveryMock::connected());
        *backend.transport.write() = Some(mock.clone() as Arc<dyn Transport>);

        backend.trip_circuit_breaker_for_test();
        assert!(backend.is_circuit_tripped(), "precondition: breaker open");

        backend
            .health_probe(Duration::from_secs(5))
            .await
            .expect("probe should succeed");

        assert!(
            !backend.is_circuit_tripped(),
            "a successful probe must reset the tripped breaker"
        );
        assert_eq!(mock.pings.load(Ordering::SeqCst), 1);
    }

    // A failing probe must NOT reset the breaker — recovery is success-gated.
    #[tokio::test]
    async fn health_probe_failure_leaves_breaker_tripped() {
        let backend = Arc::new(Backend::new(
            "test",
            BackendConfig::default(),
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        ));
        let mock = Arc::new(RecoveryMock::connected());
        mock.fail.store(true, Ordering::Relaxed);
        *backend.transport.write() = Some(mock.clone() as Arc<dyn Transport>);

        backend.trip_circuit_breaker_for_test();
        let result = backend.health_probe(Duration::from_secs(5)).await;

        assert!(result.is_err(), "failed probe returns Err");
        assert!(
            backend.is_circuit_tripped(),
            "a failed probe must leave the breaker tripped"
        );
    }

    #[test]
    fn oauth_requires_per_user_isolation_reflects_config() {
        let mk = |oauth: Option<crate::config::OAuthConfig>| {
            Backend::new(
                "b",
                BackendConfig {
                    oauth,
                    ..BackendConfig::default()
                },
                &crate::config::FailsafeConfig::default(),
                Duration::from_secs(60),
            )
        };
        let oauth = |enabled: bool, shared: bool| crate::config::OAuthConfig {
            enabled,
            scopes: vec![],
            client_id: None,
            client_secret: None,
            callback_host: None,
            callback_port: None,
            callback_path: None,
            token_refresh_buffer_secs: 300,
            shared_account: shared,
        };
        // Enabled, gateway-held, not blessed shared → guard MUST fire.
        assert!(
            mk(Some(oauth(true, false))).oauth_requires_per_user_isolation(),
            "enabled non-shared gateway-held OAuth must require per-user isolation"
        );
        // Operator blessed the account as shared → no isolation required.
        assert!(
            !mk(Some(oauth(true, true))).oauth_requires_per_user_isolation(),
            "shared_account=true opts out of the isolation guard"
        );
        // OAuth disabled → nothing to isolate.
        assert!(!mk(Some(oauth(false, false))).oauth_requires_per_user_isolation());
        // No OAuth config → nothing to isolate.
        assert!(!mk(None).oauth_requires_per_user_isolation());
    }

    // F3 sink-side guard (MIK-6746): even when Config::validate() is bypassed
    // by programmatic construction, create_oauth_client() must refuse to build a
    // backend OAuth client for a backend that also declares identity_propagation.
    // The backend OAuth would persist a gateway-held token during initialize(),
    // authenticating the transport session as the gateway before any per-request
    // per-user override — silently defeating per-user propagation. Fail closed at
    // the last chokepoint. Contradiction holds for BOTH implemented strategies.
    #[test]
    fn create_oauth_client_refuses_identity_propagation_backends() {
        let oauth_enabled = crate::config::OAuthConfig {
            enabled: true,
            scopes: vec![],
            client_id: None,
            client_secret: None,
            callback_host: None,
            callback_port: None,
            callback_path: None,
            token_refresh_buffer_secs: 300,
            shared_account: false,
        };
        let idp = |strategy: crate::identity_propagation::PropagationStrategyKind| {
            crate::identity_propagation::IdentityPropagationConfig {
                strategy,
                audience: "https://backend.example".to_string(),
                required: true,
                session_mode: crate::identity_propagation::SessionMode::Stateless,
            }
        };
        let mk = |strategy| {
            Backend::new(
                "b",
                BackendConfig {
                    oauth: Some(oauth_enabled.clone()),
                    identity_propagation: Some(idp(strategy)),
                    ..BackendConfig::default()
                },
                &crate::config::FailsafeConfig::default(),
                Duration::from_secs(60),
            )
        };
        for strategy in [
            crate::identity_propagation::PropagationStrategyKind::SignedAssertion,
            crate::identity_propagation::PropagationStrategyKind::Passthrough,
        ] {
            let backend = mk(strategy);
            match backend.create_oauth_client("https://backend.example") {
                Err(Error::ConfigValidation(_)) => {}
                Err(other) => {
                    panic!("expected ConfigValidation, got {other:?} for {strategy:?}")
                }
                Ok(_) => panic!(
                    "enabled backend oauth + identity_propagation must fail closed for {strategy:?}"
                ),
            }
        }

        // shared_account=true does NOT exempt: sharing one gateway-held token
        // still contradicts per-user propagation.
        let shared = Backend::new(
            "b",
            BackendConfig {
                oauth: Some(crate::config::OAuthConfig {
                    shared_account: true,
                    ..oauth_enabled.clone()
                }),
                identity_propagation: Some(idp(
                    crate::identity_propagation::PropagationStrategyKind::SignedAssertion,
                )),
                ..BackendConfig::default()
            },
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        );
        assert!(
            shared
                .create_oauth_client("https://backend.example")
                .is_err(),
            "shared_account=true must not exempt the F3 guard"
        );

        // No identity_propagation → enabled backend oauth proceeds (returns a
        // client), proving the guard does not over-reach.
        let plain = Backend::new(
            "b",
            BackendConfig {
                oauth: Some(oauth_enabled.clone()),
                ..BackendConfig::default()
            },
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        );
        assert!(
            plain.create_oauth_client("https://backend.example").is_ok(),
            "backend oauth without identity_propagation must still be allowed"
        );
    }

    #[test]
    fn backend_status_surfaces_ready_runtime_profile_lifecycle() {
        let cfg = BackendConfig {
            transport: TransportConfig::Stdio {
                command: "mcp-docs-server --stdio".to_string(),
                cwd: None,
                protocol_version: None,
            },
            runtime_profile: Some("containerized".to_string()),
            ..BackendConfig::default()
        };

        let mut runtime = crate::config::RuntimeConfig::default();
        runtime.availability.docker = true;
        runtime.profiles.insert(
            "containerized".to_string(),
            crate::config::RuntimeProfileConfig {
                provider: Some(crate::runtime::RuntimeProviderKind::Docker),
                image: Some("ghcr.io/example/docs-mcp:1".to_string()),
                restart: crate::runtime::RuntimeRestartPolicy {
                    max_restarts: 4,
                    backoff_secs: 11,
                },
                ..crate::config::RuntimeProfileConfig::default()
            },
        );
        let plan = runtime_plan_for_backend("docs", &cfg, &runtime).expect("runtime plan");
        let backend = Backend::new_with_runtime_plan(
            "docs",
            cfg,
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
            Some(plan),
        );

        let status = backend.status();
        let runtime = status.runtime.expect("runtime status");
        assert_eq!(runtime.profile, "containerized");
        assert_eq!(
            runtime.provider,
            crate::runtime::RuntimeProviderKind::Docker
        );
        assert_eq!(
            runtime.license_tier,
            crate::runtime::RuntimeLicenseTier::FreeCore
        );
        assert_eq!(runtime.state, BackendRuntimeState::Ready);
        assert!(runtime.denied_reasons.is_empty());
        assert!(runtime.confirmation_ids.is_empty());
        assert_eq!(runtime.restart_max_attempts, 4);
        assert_eq!(runtime.restart_backoff_secs, 11);
        assert!(runtime.health_check.contains("docker inspect"));
        assert_eq!(
            runtime.restart_command_hint.as_deref(),
            Some("docker restart mcp-gateway-docs")
        );
        assert!(runtime.rollback_step.contains("docker rm --force"));
    }

    #[test]
    fn backend_status_surfaces_confirmation_required_runtime_profile() {
        let cfg = BackendConfig {
            transport: TransportConfig::Stdio {
                command: "mcp-docs-server --stdio".to_string(),
                cwd: None,
                protocol_version: None,
            },
            runtime_profile: Some("local_privileged".to_string()),
            ..BackendConfig::default()
        };

        let mut runtime = crate::config::RuntimeConfig::default();
        runtime.profiles.insert(
            "local_privileged".to_string(),
            crate::config::RuntimeProfileConfig {
                provider: Some(crate::runtime::RuntimeProviderKind::LocalProcess),
                privileged: true,
                ..crate::config::RuntimeProfileConfig::default()
            },
        );
        let plan = runtime_plan_for_backend("docs", &cfg, &runtime).expect("runtime plan");
        let backend = Backend::new_with_runtime_plan(
            "docs",
            cfg,
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
            Some(plan),
        );

        let status = backend.status();
        let runtime = status.runtime.expect("runtime status");
        assert_eq!(runtime.profile, "local_privileged");
        assert_eq!(
            runtime.provider,
            crate::runtime::RuntimeProviderKind::LocalProcess
        );
        assert_eq!(runtime.state, BackendRuntimeState::ConfirmationRequired);
        assert!(runtime.denied_reasons.is_empty());
        assert_eq!(runtime.confirmation_ids, vec!["runtime.privileged"]);
        assert!(runtime.health_check.contains("stdio"));
        assert_eq!(
            runtime.restart_command_hint.as_deref(),
            Some("restart the gateway-managed child process")
        );
        assert!(runtime.rollback_step.contains("direct-launch"));
    }

    #[test]
    fn stdio_backend_uses_container_runtime_bridge_command() {
        let cfg = BackendConfig {
            transport: TransportConfig::Stdio {
                command: "definitely-not-a-real-mcp-server".to_string(),
                cwd: None,
                protocol_version: None,
            },
            env: HashMap::from([
                ("SAFE_HANDLE".to_string(), "safe-value".to_string()),
                ("UNDECLARED_ENV".to_string(), "must-not-pass".to_string()),
            ]),
            runtime_profile: Some("containerized".to_string()),
            ..BackendConfig::default()
        };

        let mut runtime = crate::config::RuntimeConfig::default();
        runtime.availability.docker = true;
        runtime.profiles.insert(
            "containerized".to_string(),
            crate::config::RuntimeProfileConfig {
                provider: Some(crate::runtime::RuntimeProviderKind::Docker),
                image: Some("ghcr.io/example/server:latest".to_string()),
                env_keys: vec!["SAFE_HANDLE".to_string()],
                ..crate::config::RuntimeProfileConfig::default()
            },
        );
        let plan = runtime_plan_for_backend("docs", &cfg, &runtime).expect("runtime plan");
        let backend = Backend::new_with_runtime_plan(
            "docs",
            cfg,
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
            Some(plan),
        );

        let launch = backend
            .resolve_stdio_runtime_launch("definitely-not-a-real-mcp-server")
            .expect("container stdio bridge launch");
        let parts = shlex::split(&launch.command).expect("bridge command is shell-splitable");

        assert_eq!(parts.first().map(String::as_str), Some("docker"));
        assert_eq!(parts.get(1).map(String::as_str), Some("run"));
        assert_eq!(
            parts.get(2..6),
            Some(
                &[
                    "--interactive".to_string(),
                    "--rm".to_string(),
                    "--name".to_string(),
                    "mcp-gateway-docs".to_string()
                ][..]
            ),
            "bridge flags must not split paired docker options: {parts:?}"
        );
        assert!(parts.contains(&"--interactive".to_string()));
        assert!(parts.contains(&"--rm".to_string()));
        assert!(!parts.contains(&"--detach".to_string()));
        assert!(
            !parts.iter().any(|arg| arg.starts_with("--restart=")),
            "stdio bridge must drop detached restart policy flags: {parts:?}"
        );
        assert!(parts.contains(&"--network=none".to_string()));
        assert!(parts.contains(&"--read-only".to_string()));
        assert!(parts.contains(&"--cap-drop=ALL".to_string()));
        assert!(parts.contains(&"SAFE_HANDLE".to_string()));
        assert!(!parts.contains(&"UNDECLARED_ENV".to_string()));
        assert!(parts.contains(&"ghcr.io/example/server:latest".to_string()));
        assert_eq!(
            launch.env,
            HashMap::from([("SAFE_HANDLE".to_string(), "safe-value".to_string())])
        );
    }

    #[tokio::test]
    async fn stdio_backend_requires_runtime_confirmations_before_spawn() {
        let cfg = BackendConfig {
            transport: TransportConfig::Stdio {
                command: "definitely-not-a-real-mcp-server".to_string(),
                cwd: None,
                protocol_version: None,
            },
            runtime_profile: Some("local_privileged".to_string()),
            ..BackendConfig::default()
        };

        let mut runtime = crate::config::RuntimeConfig::default();
        runtime.profiles.insert(
            "local_privileged".to_string(),
            crate::config::RuntimeProfileConfig {
                provider: Some(crate::runtime::RuntimeProviderKind::LocalProcess),
                privileged: true,
                ..crate::config::RuntimeProfileConfig::default()
            },
        );
        let plan = runtime_plan_for_backend("docs", &cfg, &runtime).expect("runtime plan");
        assert_eq!(
            plan.launch_command
                .as_ref()
                .map(|command| command.program.as_str()),
            Some("definitely-not-a-real-mcp-server")
        );
        let backend = Backend::new_with_runtime_plan(
            "docs",
            cfg,
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
            Some(plan),
        );

        let err = backend
            .start()
            .await
            .expect_err("missing runtime confirmation rejected");
        assert!(
            err.to_string().contains("requires confirmations"),
            "confirmation-required runtime plan should fail closed before spawn: {err}"
        );
    }

    fn sample_tool(name: &str) -> Tool {
        Tool {
            name: name.to_string(),
            title: None,
            description: Some(format!("{name} tool")),
            input_schema: json!({"type": "object"}),
            output_schema: None,
            annotations: None,
            role: None,
            projection: None,
        }
    }

    #[test]
    fn normalize_tool_annotations_fills_missing_hints() {
        let mut tools = vec![sample_tool("search_messages"), sample_tool("send_message")];

        normalize_tool_annotations("beeper", &mut tools);

        let search = tools[0].annotations.as_ref().unwrap();
        assert_eq!(search.read_only_hint, Some(true));
        assert_eq!(search.destructive_hint, Some(false));
        assert_eq!(search.idempotent_hint, Some(true));
        assert_eq!(search.open_world_hint, Some(true));

        let send = tools[1].annotations.as_ref().unwrap();
        assert_eq!(send.read_only_hint, Some(false));
        assert_eq!(send.destructive_hint, Some(true));
        assert_eq!(send.idempotent_hint, Some(false));
        assert_eq!(send.open_world_hint, Some(true));
    }

    #[test]
    fn normalize_tool_annotations_preserves_existing_true_hints_and_adds_false_hints() {
        let mut tool = sample_tool("recall");
        tool.annotations = Some(ToolAnnotations {
            read_only_hint: Some(true),
            destructive_hint: None,
            idempotent_hint: None,
            open_world_hint: None,
            title: None,
        });
        let mut tools = vec![tool];

        normalize_tool_annotations("hebb", &mut tools);

        let annotations = tools[0].annotations.as_ref().unwrap();
        assert_eq!(annotations.read_only_hint, Some(true));
        assert_eq!(annotations.destructive_hint, Some(false));
        assert_eq!(annotations.idempotent_hint, Some(true));
        assert_eq!(annotations.open_world_hint, Some(false));
    }

    #[test]
    fn normalize_tool_annotations_preserves_downstream_annotation_title_and_hints() {
        let mut tool = sample_tool("remote_write");
        tool.annotations = Some(ToolAnnotations {
            title: Some("Remote Write".to_string()),
            read_only_hint: Some(false),
            destructive_hint: Some(false),
            idempotent_hint: Some(false),
            open_world_hint: Some(false),
        });
        let mut tools = vec![tool];

        normalize_tool_annotations("remote-api", &mut tools);

        let annotations = tools[0].annotations.as_ref().unwrap();
        assert_eq!(annotations.title.as_deref(), Some("Remote Write"));
        assert_eq!(annotations.read_only_hint, Some(false));
        assert_eq!(annotations.destructive_hint, Some(false));
        assert_eq!(annotations.idempotent_hint, Some(false));
        assert_eq!(annotations.open_world_hint, Some(false));
    }

    #[test]
    fn cached_metadata_tracks_freshness() {
        let cache = CachedMetadata::new();
        assert!(!cache.is_fresh(Duration::from_secs(60)));

        cache.store_shared(Arc::new(vec![1, 2, 3]));

        assert!(cache.is_fresh(Duration::from_secs(60)));
        let snapshot = cache.snapshot_shared().unwrap();
        assert_eq!(snapshot.as_ref(), &vec![1, 2, 3]);
        assert_eq!(snapshot.len(), 3);
    }

    #[tokio::test]
    async fn cached_metadata_shared_reads_reuse_arc() {
        let cache = CachedMetadata::new();

        let first = cache
            .get_or_fetch_shared(Duration::from_secs(60), || async { Ok(vec![1, 2, 3]) })
            .await
            .unwrap();
        let second = cache
            .get_or_fetch_shared(Duration::from_secs(60), || async {
                panic!("fresh cache hit should not refetch")
            })
            .await
            .unwrap();

        assert!(Arc::ptr_eq(&first, &second));
    }

    #[tokio::test]
    async fn cached_metadata_retries_after_fetch_error() {
        let cache = CachedMetadata::new();
        let attempts = AtomicUsize::new(0);

        let first = cache
            .get_or_fetch_shared(Duration::from_secs(60), || async {
                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
                if attempt == 0 {
                    Err(Error::BackendUnavailable("boom".to_string()))
                } else {
                    Ok(vec![7])
                }
            })
            .await;
        assert!(first.is_err());

        let second = cache
            .get_or_fetch_shared(Duration::from_secs(60), || async {
                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
                if attempt == 0 {
                    Err(Error::BackendUnavailable("boom".to_string()))
                } else {
                    Ok(vec![7])
                }
            })
            .await;

        assert_eq!(second.unwrap().as_ref(), &vec![7]);
        assert_eq!(attempts.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn get_tools_singleflight_coalesces_concurrent_requests() {
        let backend = Arc::new(Backend::new(
            "test",
            BackendConfig::default(),
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        ));
        let response = JsonRpcResponse::success_serialized(
            RequestId::Number(1),
            ToolsListResult {
                tools: vec![sample_tool("echo")],
                next_cursor: None,
            },
        );
        let transport = Arc::new(MockTransport::new(response, Duration::from_millis(25)));
        let transport_dyn: Arc<dyn Transport> = transport.clone();
        *backend.transport.write() = Some(transport_dyn);

        let barrier = Arc::new(Barrier::new(6));
        let mut tasks = Vec::new();
        for _ in 0..5 {
            let backend = Arc::clone(&backend);
            let barrier = Arc::clone(&barrier);
            tasks.push(tokio::spawn(async move {
                barrier.wait().await;
                backend.get_tools().await.unwrap()
            }));
        }

        barrier.wait().await;

        for task in tasks {
            let tools = task.await.unwrap();
            assert_eq!(tools.len(), 1);
            assert_eq!(tools[0].name, "echo");
        }

        assert_eq!(transport.requests.load(Ordering::SeqCst), 1);
        assert!(backend.has_cached_tools());
        assert_eq!(backend.cached_tools_count(), 1);
        assert_eq!(
            backend.get_cached_tool("echo").map(|tool| tool.name),
            Some("echo".to_string())
        );
    }

    #[tokio::test]
    async fn get_tools_does_not_cache_json_rpc_error_response() {
        let backend = Arc::new(Backend::new(
            "test",
            BackendConfig::default(),
            &crate::config::FailsafeConfig::default(),
            Duration::from_secs(60),
        ));
        let response = JsonRpcResponse::error(Some(RequestId::Number(1)), -32000, "backend down");
        let transport = Arc::new(MockTransport::new(response, Duration::from_millis(0)));
        let transport_dyn: Arc<dyn Transport> = transport.clone();
        *backend.transport.write() = Some(transport_dyn);

        let result = backend.get_tools().await;

        assert!(result.is_err());
        assert!(!backend.has_cached_tools());
        assert_eq!(transport.requests.load(Ordering::SeqCst), 1);
    }
}
