agent-infra-sdk 0.2.1

use infra_api_gateway_contract::{
    AUDIT_PATH, CREDENTIAL_EXCHANGE_PATH, CredentialExchangeRequest, CredentialExchangeResponse,
    DEPENDENCIES_PATH, DISCOVERY_PATH, DiscoveryDocument, EFFECTIVE_POLICY_PATH,
    EXECUTION_DELEGATION_CREDENTIAL_PATH, EXECUTION_DELEGATION_RENEW_PATH,
    EXECUTION_DELEGATION_REVOKE_PATH, EXECUTION_DELEGATIONS_PATH, EffectivePolicy,
    EstablishExecutionDelegationRequest, ExecutionDelegationLease, GatewayAuditPage,
    GatewayMetricsSnapshot, GatewayStatus, HealthSnapshot, LIVE_PATH, METRICS_PATH,
    MintExecutionDelegationCredentialRequest, READY_PATH, ROUTES_PATH,
    RenewExecutionDelegationRequest, RevokeExecutionDelegationRequest, RouteSnapshot,
    RUNTIME_CREDENTIAL_BOOTSTRAP_PATH,
};
#[cfg(feature = "runtime-identity")]
use agent_runtime_identity_contract::{
    IssueRuntimeCredentialRequest, IssueRuntimeCredentialResponse,
};
use reqwest::Client;
use std::collections::BTreeMap;
use std::fmt;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock};

use crate::transport::{
    BearerCredential, CallOptions, ClientOptions, CredentialError, CredentialsProvider,
    HttpTransport, InfraClientError, ServiceEndpoint,
};
use crate::transport_support::secure_transport;

#[derive(Clone, Debug)]
pub struct GatewayClient {
    transport: HttpTransport,
}

#[derive(Clone)]
struct CachedCredential {
    value: BearerCredential,
    refresh_at: Instant,
}

struct CredentialSlot {
    capabilities: Vec<String>,
    cached: RwLock<Option<CachedCredential>>,
    refresh: Mutex<()>,
}

/// Exchanges one external workload credential for short-lived,
/// audience-bound service credentials. Cache growth is bounded by the
/// explicitly configured audience map, and refresh is single-flight.
#[derive(Clone)]
pub struct GatewayExchangeCredentials {
    http: Client,
    exchange_url: Arc<str>,
    source: Arc<dyn CredentialsProvider>,
    slots: Arc<BTreeMap<String, CredentialSlot>>,
    timeout: Duration,
}

/// Mints short-lived service credentials from an opaque execution delegation
/// lease. The lease reference is never exposed through `Debug`, and the
/// workload credential remains owned by the configured source provider.
#[derive(Clone)]
pub struct DelegationLeaseCredentials {
    http: Client,
    mint_url: Arc<str>,
    lease_ref: Arc<str>,
    source: Arc<dyn CredentialsProvider>,
    slots: Arc<BTreeMap<String, CredentialSlot>>,
    timeout: Duration,
}

impl fmt::Debug for DelegationLeaseCredentials {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DelegationLeaseCredentials")
            .field("mint_url", &self.mint_url)
            .field("lease_ref", &"[REDACTED OPAQUE REFERENCE]")
            .field("audiences", &self.slots.keys().collect::<Vec<_>>())
            .field("source", &"[REDACTED CREDENTIAL PROVIDER]")
            .finish_non_exhaustive()
    }
}

impl DelegationLeaseCredentials {
    pub fn new(
        gateway_base_url: impl Into<String>,
        lease_ref: impl Into<Arc<str>>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
    ) -> Result<Self, CredentialError> {
        Self::new_with_transport_policy(gateway_base_url, lease_ref, source, capabilities, false)
    }

    /// Use only when an authenticated service mesh protects plaintext traffic
    /// between this workload and the gateway.
    pub fn new_trusted_mesh_http(
        gateway_base_url: impl Into<String>,
        lease_ref: impl Into<Arc<str>>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
    ) -> Result<Self, CredentialError> {
        Self::new_with_transport_policy(gateway_base_url, lease_ref, source, capabilities, true)
    }

    fn new_with_transport_policy(
        gateway_base_url: impl Into<String>,
        lease_ref: impl Into<Arc<str>>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
        trusted_mesh_http: bool,
    ) -> Result<Self, CredentialError> {
        let lease_ref = lease_ref.into();
        if !lease_ref.starts_with("edl_")
            || lease_ref.len() > 96
            || lease_ref
                .bytes()
                .any(|byte| !(byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')))
        {
            return Err(CredentialError(
                "execution delegation reference is invalid".into(),
            ));
        }
        validate_credential_capabilities(&capabilities)?;
        let mint_url = gateway_credential_url(
            gateway_base_url.into(),
            EXECUTION_DELEGATION_CREDENTIAL_PATH,
            trusted_mesh_http,
        )?;
        Ok(Self {
            http: credential_http_client()?,
            mint_url: mint_url.into(),
            lease_ref,
            source,
            slots: Arc::new(credential_slots(capabilities)),
            timeout: Duration::from_secs(5),
        })
    }
}

impl fmt::Debug for GatewayExchangeCredentials {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("GatewayExchangeCredentials")
            .field("exchange_url", &self.exchange_url)
            .field("audiences", &self.slots.keys().collect::<Vec<_>>())
            .field("source", &"[REDACTED CREDENTIAL PROVIDER]")
            .finish_non_exhaustive()
    }
}

impl GatewayExchangeCredentials {
    pub fn new(
        gateway_base_url: impl Into<String>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
    ) -> Result<Self, CredentialError> {
        Self::new_with_transport_policy(gateway_base_url, source, capabilities, false)
    }

    /// Use only when an authenticated service mesh protects plaintext traffic
    /// between this workload and the gateway.
    pub fn new_trusted_mesh_http(
        gateway_base_url: impl Into<String>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
    ) -> Result<Self, CredentialError> {
        Self::new_with_transport_policy(gateway_base_url, source, capabilities, true)
    }

    fn new_with_transport_policy(
        gateway_base_url: impl Into<String>,
        source: Arc<dyn CredentialsProvider>,
        capabilities: BTreeMap<String, Vec<String>>,
        trusted_mesh_http: bool,
    ) -> Result<Self, CredentialError> {
        validate_credential_capabilities(&capabilities)?;
        let exchange_url = gateway_credential_url(
            gateway_base_url.into(),
            CREDENTIAL_EXCHANGE_PATH,
            trusted_mesh_http,
        )?;
        Ok(Self {
            http: credential_http_client()?,
            exchange_url: exchange_url.into(),
            source,
            slots: Arc::new(credential_slots(capabilities)),
            timeout: Duration::from_secs(5),
        })
    }

    fn fresh(cached: &CachedCredential) -> bool {
        Instant::now() < cached.refresh_at
    }

    async fn cached(slot: &CredentialSlot) -> Option<BearerCredential> {
        slot.cached
            .read()
            .await
            .as_ref()
            .filter(|cached| Self::fresh(cached))
            .map(|cached| cached.value.clone())
    }
}

fn validate_credential_capabilities(
    capabilities: &BTreeMap<String, Vec<String>>,
) -> Result<(), CredentialError> {
    if capabilities.is_empty() || capabilities.len() > 32 {
        return Err(CredentialError(
            "credential provider requires 1..=32 configured audiences".into(),
        ));
    }
    if capabilities.iter().any(|(audience, values)| {
        audience.is_empty()
            || audience.len() > 128
            || values.is_empty()
            || values.len() > 32
            || values
                .iter()
                .any(|value| value.is_empty() || value.len() > 128)
    }) {
        return Err(CredentialError(
            "credential provider audience or capabilities are invalid".into(),
        ));
    }
    Ok(())
}

fn gateway_credential_url(
    base: String,
    path: &str,
    trusted_mesh_http: bool,
) -> Result<String, CredentialError> {
    let mut base_url = reqwest::Url::parse(&base)
        .map_err(|_| CredentialError("gateway credential endpoint is invalid".into()))?;
    if base_url.host_str().is_none()
        || !secure_transport(&base_url, trusted_mesh_http)
        || !base_url.username().is_empty()
        || base_url.password().is_some()
        || base_url.query().is_some()
        || base_url.fragment().is_some()
    {
        return Err(CredentialError(
            "gateway credential endpoint must use HTTPS, loopback HTTP, or explicitly trusted mesh HTTP without embedded credentials".into(),
        ));
    }
    base_url.set_path(&format!("{}/", base_url.path().trim_end_matches('/')));
    base_url
        .join(path.trim_start_matches('/'))
        .map(|url| url.to_string())
        .map_err(|_| CredentialError("gateway credential endpoint is invalid".into()))
}

fn credential_http_client() -> Result<Client, CredentialError> {
    Client::builder()
        .connect_timeout(Duration::from_secs(3))
        .timeout(Duration::from_secs(5))
        .redirect(reqwest::redirect::Policy::none())
        .pool_idle_timeout(Duration::from_secs(60))
        .pool_max_idle_per_host(4)
        .build()
        .map_err(|_| CredentialError("failed to build gateway credential client".into()))
}

fn credential_slots(
    capabilities: BTreeMap<String, Vec<String>>,
) -> BTreeMap<String, CredentialSlot> {
    capabilities
        .into_iter()
        .map(|(audience, capabilities)| {
            (
                audience,
                CredentialSlot {
                    capabilities,
                    cached: RwLock::new(None),
                    refresh: Mutex::new(()),
                },
            )
        })
        .collect()
}

#[async_trait::async_trait]
impl CredentialsProvider for GatewayExchangeCredentials {
    async fn credential(&self, audience: &str) -> Result<BearerCredential, CredentialError> {
        let slot = self
            .slots
            .get(audience)
            .ok_or_else(|| CredentialError(format!("audience {audience} is not configured")))?;
        let deadline = Instant::now() + self.timeout;
        if let Some(value) = Self::cached(slot).await {
            return Ok(value);
        }
        let _refresh = tokio::time::timeout(
            deadline.saturating_duration_since(Instant::now()),
            slot.refresh.lock(),
        )
        .await
        .map_err(|_| CredentialError("credential refresh single-flight timed out".into()))?;
        if let Some(value) = Self::cached(slot).await {
            return Ok(value);
        }
        let request = CredentialExchangeRequest {
            audience: audience.to_string(),
            capabilities: slot.capabilities.clone(),
        };
        let source = tokio::time::timeout(
            deadline.saturating_duration_since(Instant::now()),
            self.source.credential("agent-infra"),
        )
        .await
        .map_err(|_| CredentialError("source credential acquisition timed out".into()))??;
        let mut attempt = 0_usize;
        let response = loop {
            let remaining = deadline
                .checked_duration_since(Instant::now())
                .ok_or_else(|| CredentialError("gateway credential exchange timed out".into()))?;
            let result = self
                .http
                .post(self.exchange_url.as_ref())
                .timeout(remaining)
                .bearer_auth(source.expose())
                .json(&request)
                .send()
                .await;
            match result {
                Ok(response)
                    if attempt < 2 && retryable_exchange_status(response.status().as_u16()) =>
                {
                    attempt += 1;
                }
                Ok(response) => break response,
                Err(error) if attempt < 2 && (error.is_connect() || error.is_timeout()) => {
                    attempt += 1;
                }
                Err(_) => {
                    return Err(CredentialError("gateway credential exchange failed".into()));
                }
            }
            let delay = Duration::from_millis(25 * attempt as u64);
            if delay >= deadline.saturating_duration_since(Instant::now()) {
                return Err(CredentialError(
                    "gateway credential exchange timed out".into(),
                ));
            }
            tokio::time::sleep(delay).await;
        };
        let (value, expires_in_seconds) = decode_gateway_credential(response).await?;
        let refresh_at = credential_refresh_at(expires_in_seconds);
        *slot.cached.write().await = Some(CachedCredential {
            value: value.clone(),
            refresh_at,
        });
        Ok(value)
    }
}

#[async_trait::async_trait]
impl CredentialsProvider for DelegationLeaseCredentials {
    async fn credential(&self, audience: &str) -> Result<BearerCredential, CredentialError> {
        let slot = self
            .slots
            .get(audience)
            .ok_or_else(|| CredentialError(format!("audience {audience} is not configured")))?;
        let deadline = Instant::now() + self.timeout;
        if let Some(value) = GatewayExchangeCredentials::cached(slot).await {
            return Ok(value);
        }
        let _refresh = tokio::time::timeout(
            deadline.saturating_duration_since(Instant::now()),
            slot.refresh.lock(),
        )
        .await
        .map_err(|_| CredentialError("credential refresh single-flight timed out".into()))?;
        if let Some(value) = GatewayExchangeCredentials::cached(slot).await {
            return Ok(value);
        }
        let source = tokio::time::timeout(
            deadline.saturating_duration_since(Instant::now()),
            self.source.credential("agent-infra"),
        )
        .await
        .map_err(|_| CredentialError("source credential acquisition timed out".into()))??;
        let request = MintExecutionDelegationCredentialRequest {
            lease_ref: self.lease_ref.to_string(),
            audience: audience.to_string(),
            capabilities: slot.capabilities.clone(),
        };
        let mut attempt = 0_usize;
        let response = loop {
            let remaining = deadline
                .checked_duration_since(Instant::now())
                .ok_or_else(|| CredentialError("delegation credential mint timed out".into()))?;
            let result = self
                .http
                .post(self.mint_url.as_ref())
                .timeout(remaining)
                .bearer_auth(source.expose())
                .json(&request)
                .send()
                .await;
            match result {
                Ok(response)
                    if attempt < 2 && retryable_exchange_status(response.status().as_u16()) =>
                {
                    attempt += 1;
                }
                Ok(response) => break response,
                Err(error) if attempt < 2 && (error.is_connect() || error.is_timeout()) => {
                    attempt += 1;
                }
                Err(_) => {
                    return Err(CredentialError("delegation credential mint failed".into()));
                }
            }
            let delay = Duration::from_millis(25 * attempt as u64);
            if delay >= deadline.saturating_duration_since(Instant::now()) {
                return Err(CredentialError(
                    "delegation credential mint timed out".into(),
                ));
            }
            tokio::time::sleep(delay).await;
        };
        let (value, expires_in_seconds) = decode_gateway_credential(response).await?;
        *slot.cached.write().await = Some(CachedCredential {
            value: value.clone(),
            refresh_at: credential_refresh_at(expires_in_seconds),
        });
        Ok(value)
    }
}

async fn decode_gateway_credential(
    mut response: reqwest::Response,
) -> Result<(BearerCredential, u64), CredentialError> {
    if !response.status().is_success()
        || response
            .content_length()
            .is_some_and(|length| length > 64 * 1024)
    {
        return Err(CredentialError(format!(
            "gateway credential request returned HTTP {}",
            response.status().as_u16()
        )));
    }
    let mut bytes =
        Vec::with_capacity(response.content_length().unwrap_or(0).min(64 * 1024) as usize);
    while let Some(chunk) = response
        .chunk()
        .await
        .map_err(|_| CredentialError("gateway credential response read failed".into()))?
    {
        if bytes.len().saturating_add(chunk.len()) > 64 * 1024 {
            return Err(CredentialError(
                "gateway credential response exceeded 64 KiB".into(),
            ));
        }
        bytes.extend_from_slice(&chunk);
    }
    let exchanged: CredentialExchangeResponse = serde_json::from_slice(&bytes)
        .map_err(|_| CredentialError("gateway credential response was invalid".into()))?;
    if exchanged.token_type != "Bearer" || !(30..=3_600).contains(&exchanged.expires_in_seconds) {
        return Err(CredentialError(
            "gateway returned an unusable credential".into(),
        ));
    }
    Ok((
        BearerCredential::new(exchanged.access_token)?,
        exchanged.expires_in_seconds,
    ))
}

fn credential_refresh_at(expires_in_seconds: u64) -> Instant {
    let refresh_margin = (expires_in_seconds / 5).clamp(5, 30);
    Instant::now() + Duration::from_secs(expires_in_seconds.saturating_sub(refresh_margin))
}

fn retryable_exchange_status(status: u16) -> bool {
    matches!(status, 408 | 429 | 502 | 503 | 504)
}

impl GatewayClient {
    pub(crate) fn new_with_endpoint(
        http: Client,
        endpoint: ServiceEndpoint,
        options: ClientOptions,
    ) -> Self {
        let endpoint = endpoint.with_default_credential_audience("agent-infra");
        Self {
            transport: HttpTransport::new_with_options(http, "gateway", endpoint, options),
        }
    }

    pub async fn discovery(&self) -> Result<DiscoveryDocument, InfraClientError> {
        self.transport.get_json(DISCOVERY_PATH).await
    }

    /// Gateway liveness. `Ok(())` means the gateway process is accepting
    /// requests; it says nothing about downstream dependencies.
    pub async fn live(&self) -> Result<GatewayStatus, InfraClientError> {
        self.transport.get_json(LIVE_PATH).await
    }

    /// Gateway readiness. Returns the readiness body when the gateway and its
    /// control-plane dependencies are ready. A `503` readiness response is
    /// reported as `Ok(status)` with `status.is_ready() == false` rather than
    /// as an error, so callers can gate startup on it.
    pub async fn ready(&self) -> Result<GatewayStatus, InfraClientError> {
        match self.transport.get_json::<GatewayStatus>(READY_PATH).await {
            Ok(status) => Ok(status),
            Err(InfraClientError::HttpStatus { status: 503, .. }) => Ok(GatewayStatus {
                status: "not_ready".to_string(),
                service: "infra-api-gateway".to_string(),
            }),
            Err(error) => Err(error),
        }
    }

    /// Poll [`ready`](Self::ready) until the gateway reports ready or `timeout`
    /// elapses. Transport errors (unreachable gateway, TLS failures) are
    /// retried within the budget; a `not_ready` gateway is retried until it
    /// becomes ready. Returns [`InfraClientError::DeadlineExceeded`] if the
    /// gateway is still not ready when the timeout is reached.
    pub async fn wait_until_ready(
        &self,
        timeout: Duration,
    ) -> Result<GatewayStatus, InfraClientError> {
        const POLL_INTERVAL: Duration = Duration::from_millis(250);
        let deadline = Instant::now() + timeout;
        loop {
            match self.ready().await {
                Ok(status) if status.is_ready() => return Ok(status),
                Ok(_) => {}
                Err(error) if Instant::now() >= deadline => return Err(error),
                Err(_) => {}
            }
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(InfraClientError::DeadlineExceeded { service: "gateway" });
            }
            tokio::time::sleep(POLL_INTERVAL.min(remaining)).await;
        }
    }

    pub async fn dependencies(&self) -> Result<HealthSnapshot, InfraClientError> {
        self.transport.get_json(DEPENDENCIES_PATH).await
    }

    pub async fn routes(&self) -> Result<RouteSnapshot, InfraClientError> {
        self.transport.get_json(ROUTES_PATH).await
    }

    pub async fn metrics(&self) -> Result<GatewayMetricsSnapshot, InfraClientError> {
        self.transport.get_json(METRICS_PATH).await
    }

    pub async fn effective_policy(&self) -> Result<EffectivePolicy, InfraClientError> {
        self.transport.get_json(EFFECTIVE_POLICY_PATH).await
    }

    pub async fn exchange_credential(
        &self,
        request: &CredentialExchangeRequest,
    ) -> Result<CredentialExchangeResponse, InfraClientError> {
        self.transport
            .post_json(CREDENTIAL_EXCHANGE_PATH, request)
            .await
    }

    /// Exchange a single-use Runtime Identity bootstrap proof for the first
    /// Runtime workload credential. This call intentionally sends no Gateway
    /// bearer; the proof inside `request` is validated by Runtime Identity.
    #[cfg(feature = "runtime-identity")]
    pub async fn bootstrap_runtime_credential(
        &self,
        request: &IssueRuntimeCredentialRequest,
    ) -> Result<IssueRuntimeCredentialResponse, InfraClientError> {
        self.transport
            .post_json_without_credentials(RUNTIME_CREDENTIAL_BOOTSTRAP_PATH, request)
            .await
    }

    /// Exchange a transient caller bearer plus the configured Runtime workload
    /// credential for an opaque, execution-bound delegation reference.
    pub async fn establish_execution_delegation(
        &self,
        caller: BearerCredential,
        request: &EstablishExecutionDelegationRequest,
        options: CallOptions,
    ) -> Result<ExecutionDelegationLease, InfraClientError> {
        self.transport
            .post_json_with_options(
                EXECUTION_DELEGATIONS_PATH,
                request,
                options
                    .caller_credential(caller)
                    .idempotency_key(&request.idempotency_key),
            )
            .await
    }

    pub async fn mint_execution_delegation_credential(
        &self,
        request: &MintExecutionDelegationCredentialRequest,
        options: CallOptions,
    ) -> Result<CredentialExchangeResponse, InfraClientError> {
        self.transport
            .post_json_with_options(EXECUTION_DELEGATION_CREDENTIAL_PATH, request, options)
            .await
    }

    pub async fn renew_execution_delegation(
        &self,
        request: &RenewExecutionDelegationRequest,
        options: CallOptions,
    ) -> Result<ExecutionDelegationLease, InfraClientError> {
        self.transport
            .post_json_with_options(
                EXECUTION_DELEGATION_RENEW_PATH,
                request,
                options.idempotency_key(&request.idempotency_key),
            )
            .await
    }

    pub async fn revoke_execution_delegation(
        &self,
        request: &RevokeExecutionDelegationRequest,
        options: CallOptions,
    ) -> Result<(), InfraClientError> {
        self.transport
            .post_json_with_options(
                EXECUTION_DELEGATION_REVOKE_PATH,
                request,
                options.idempotency_key(&request.idempotency_key),
            )
            .await
    }

    pub async fn audit_page(
        &self,
        cursor: Option<&str>,
        limit: Option<usize>,
    ) -> Result<GatewayAuditPage, InfraClientError> {
        let mut path = AUDIT_PATH.to_string();
        let mut query = Vec::new();
        if let Some(cursor) = cursor {
            query.push(format!("cursor={}", encode_query(cursor)));
        }
        if let Some(limit) = limit {
            query.push(format!("limit={}", limit.min(500)));
        }
        if !query.is_empty() {
            path.push('?');
            path.push_str(&query.join("&"));
        }
        self.transport.get_json(&path).await
    }

    pub async fn update_routes(
        &self,
        snapshot: &RouteSnapshot,
    ) -> Result<RouteSnapshot, InfraClientError> {
        self.transport
            .put_json_with_options(
                ROUTES_PATH,
                snapshot,
                CallOptions::default()
                    .idempotency_key(format!("gateway-routes:{}", snapshot.version)),
            )
            .await
    }
}

fn encode_query(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
            encoded.push(char::from(byte));
        } else {
            use std::fmt::Write as _;
            write!(&mut encoded, "%{byte:02X}").expect("writing to String cannot fail");
        }
    }
    encoded
}

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

    #[derive(Debug)]
    struct HangingCredentials;

    #[async_trait::async_trait]
    impl CredentialsProvider for HangingCredentials {
        async fn credential(&self, _audience: &str) -> Result<BearerCredential, CredentialError> {
            std::future::pending().await
        }
    }

    #[tokio::test]
    async fn exchange_deadline_includes_source_credential_acquisition() {
        let mut provider = GatewayExchangeCredentials::new(
            "http://127.0.0.1:1",
            Arc::new(HangingCredentials),
            BTreeMap::from([("agent-context-infra".into(), vec!["messages:read".into()])]),
        )
        .unwrap();
        provider.timeout = Duration::from_millis(20);
        let error = provider
            .credential("agent-context-infra")
            .await
            .unwrap_err();
        assert_eq!(error.code(), "CREDENTIAL_TIMEOUT");
    }

    #[test]
    fn exchange_source_credential_requires_a_secure_first_hop() {
        let capabilities =
            BTreeMap::from([("agent-context-infra".into(), vec!["messages:read".into()])]);
        assert!(
            GatewayExchangeCredentials::new(
                "http://gateway.service:5200",
                Arc::new(HangingCredentials),
                capabilities.clone(),
            )
            .is_err()
        );
        assert!(
            GatewayExchangeCredentials::new_trusted_mesh_http(
                "http://gateway.service:5200",
                Arc::new(HangingCredentials),
                capabilities,
            )
            .is_ok()
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore = "manual release microbenchmark; run with --release --ignored --nocapture"]
    async fn benchmark_global_vs_per_audience_credential_refresh_lock() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use tokio::sync::Mutex;

        async fn sample(
            locks: Arc<Vec<Arc<Mutex<()>>>>,
            calls: Arc<AtomicUsize>,
            active: Arc<AtomicUsize>,
            peak: Arc<AtomicUsize>,
        ) -> u128 {
            let started = std::time::Instant::now();
            let mut tasks = tokio::task::JoinSet::new();
            for audience in 0..4 {
                let locks = locks.clone();
                let calls = calls.clone();
                let active = active.clone();
                let peak = peak.clone();
                tasks.spawn(async move {
                    let _guard = locks[audience].lock().await;
                    calls.fetch_add(1, Ordering::Relaxed);
                    let concurrent = active.fetch_add(1, Ordering::SeqCst) + 1;
                    peak.fetch_max(concurrent, Ordering::SeqCst);
                    tokio::time::sleep(Duration::from_millis(5)).await;
                    active.fetch_sub(1, Ordering::SeqCst);
                });
            }
            while let Some(result) = tasks.join_next().await {
                result.unwrap();
            }
            started.elapsed().as_nanos()
        }

        fn summarize(label: &str, mut samples: Vec<u128>, calls: usize, peak: usize) {
            samples.sort_unstable();
            let mean = samples.iter().sum::<u128>() / samples.len() as u128;
            let variance = samples
                .iter()
                .map(|value| value.abs_diff(mean).saturating_pow(2))
                .sum::<u128>()
                / samples.len() as u128;
            let percentile = |percent: usize| samples[samples.len() * percent / 100];
            eprintln!(
                "credential_refresh_{label} ns: n={} audiences=4 mock_latency_ms=5 mean={} p50={} p95={} p99={} variance={} calls={} peak_slots={}",
                samples.len(),
                mean,
                percentile(50),
                percentile(95),
                percentile(99),
                variance,
                calls,
                peak
            );
        }

        let mut before = Vec::with_capacity(30);
        let before_calls = Arc::new(AtomicUsize::new(0));
        let before_active = Arc::new(AtomicUsize::new(0));
        let before_peak = Arc::new(AtomicUsize::new(0));
        let global = Arc::new(Mutex::new(()));
        let global_locks = Arc::new(vec![global.clone(), global.clone(), global.clone(), global]);
        for _ in 0..30 {
            before.push(
                sample(
                    global_locks.clone(),
                    before_calls.clone(),
                    before_active.clone(),
                    before_peak.clone(),
                )
                .await,
            );
        }

        let mut after = Vec::with_capacity(30);
        let after_calls = Arc::new(AtomicUsize::new(0));
        let after_active = Arc::new(AtomicUsize::new(0));
        let after_peak = Arc::new(AtomicUsize::new(0));
        let audience_locks: Arc<Vec<Arc<Mutex<()>>>> =
            Arc::new((0..4).map(|_| Arc::new(Mutex::new(()))).collect());
        for _ in 0..30 {
            after.push(
                sample(
                    audience_locks.clone(),
                    after_calls.clone(),
                    after_active.clone(),
                    after_peak.clone(),
                )
                .await,
            );
        }
        summarize(
            "before",
            before,
            before_calls.load(Ordering::Relaxed),
            before_peak.load(Ordering::Relaxed),
        );
        summarize(
            "after",
            after,
            after_calls.load(Ordering::Relaxed),
            after_peak.load(Ordering::Relaxed),
        );
    }
}