agent-infra-sdk 0.1.1

use async_trait::async_trait;
use reqwest::{Client, Method, Url};
use serde::{Serialize, de::DeserializeOwned};
use std::error::Error;
use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::transport_support::{
    elapsed_micros, error_envelope as parse_error_envelope, new_traceparent, parse_retry_after,
    redacted_endpoint, secure_transport, valid_traceparent,
};

const DEFAULT_MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
const MAX_ERROR_MESSAGE_BYTES: usize = 16 * 1024;
static RETRY_JITTER_SEQUENCE: AtomicU64 = AtomicU64::new(1);

#[derive(Clone)]
pub struct ServiceEndpoint {
    pub base_url: String,
    pub bearer_token: Option<String>,
    pub credentials: Option<Arc<dyn CredentialsProvider>>,
    pub credential_audience: Option<String>,
}

impl ServiceEndpoint {
    pub fn new(base_url: impl Into<String>) -> Self {
        Self {
            base_url: base_url.into(),
            bearer_token: None,
            credentials: None,
            credential_audience: None,
        }
    }

    #[cfg(test)]
    pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
        self.bearer_token = Some(token.into());
        self
    }

    #[cfg(test)]
    pub fn with_credentials(mut self, provider: Arc<dyn CredentialsProvider>) -> Self {
        self.credentials = Some(provider);
        self
    }

    #[cfg(test)]
    pub fn with_credential_audience(mut self, audience: impl Into<String>) -> Self {
        self.credential_audience = Some(audience.into());
        self
    }

    #[allow(dead_code)] // Individual SDK feature builds use different typed facades.
    pub(crate) fn with_default_credential_audience(mut self, audience: &str) -> Self {
        if self.credential_audience.is_none() {
            self.credential_audience = Some(audience.to_string());
        }
        self
    }
}

impl fmt::Debug for ServiceEndpoint {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ServiceEndpoint")
            .field("base_url", &redacted_endpoint(&self.base_url))
            .field("bearer_token_configured", &self.bearer_token.is_some())
            .field(
                "credentials_provider_configured",
                &self.credentials.is_some(),
            )
            .field("credential_audience", &self.credential_audience)
            .finish()
    }
}

#[derive(Clone)]
pub struct BearerCredential(Arc<str>);

impl BearerCredential {
    pub fn new(token: impl Into<Arc<str>>) -> Result<Self, CredentialError> {
        let token = token.into();
        if token.is_empty() || token.bytes().any(|byte| byte.is_ascii_whitespace()) {
            return Err(CredentialError(
                "credential token is empty or malformed".into(),
            ));
        }
        Ok(Self(token))
    }

    pub(crate) fn expose(&self) -> &str {
        &self.0
    }
}

impl fmt::Debug for BearerCredential {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("BearerCredential([REDACTED])")
    }
}

#[derive(Clone)]
pub struct CredentialError(pub String);

impl CredentialError {
    pub fn code(&self) -> &'static str {
        if self.0.contains("timed out") || self.0.contains("deadline") {
            "CREDENTIAL_TIMEOUT"
        } else {
            "CREDENTIAL"
        }
    }
}

impl fmt::Debug for CredentialError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("CredentialError([REDACTED])")
    }
}

impl fmt::Display for CredentialError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("credential provider failed ([REDACTED])")
    }
}

impl Error for CredentialError {}

#[async_trait]
pub trait CredentialsProvider: Send + Sync + fmt::Debug {
    async fn credential(&self, audience: &str) -> Result<BearerCredential, CredentialError>;
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TelemetryPhase {
    Start,
    Attempt,
    Retry,
    Result,
}

#[derive(Debug, Clone, Copy)]
pub struct TelemetryEvent {
    pub service: &'static str,
    pub phase: TelemetryPhase,
    pub attempt: usize,
    pub status: Option<u16>,
    pub elapsed_micros: u64,
    pub response_bytes: Option<usize>,
    pub retry_delay_millis: Option<u64>,
    pub outcome: &'static str,
}

pub trait TelemetryObserver: Send + Sync + fmt::Debug {
    /// Receives bounded metadata only. Request/response bodies, URLs and
    /// credentials are deliberately absent from this event contract.
    fn observe(&self, event: TelemetryEvent);
}

#[derive(Debug, Default)]
pub struct NoopTelemetry;

impl TelemetryObserver for NoopTelemetry {
    fn observe(&self, _event: TelemetryEvent) {}
}

#[derive(Debug, Clone)]
pub struct StaticCredentials(BearerCredential);

impl StaticCredentials {
    pub fn new(token: impl Into<Arc<str>>) -> Result<Self, CredentialError> {
        Ok(Self(BearerCredential::new(token)?))
    }
}

#[async_trait]
impl CredentialsProvider for StaticCredentials {
    async fn credential(&self, _audience: &str) -> Result<BearerCredential, CredentialError> {
        Ok(self.0.clone())
    }
}

#[derive(Debug, Clone)]
pub struct RetryPolicy {
    /// Total attempts, including the first request.
    pub max_attempts: usize,
    pub base_delay: Duration,
    pub max_delay: Duration,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            base_delay: Duration::from_millis(50),
            max_delay: Duration::from_secs(2),
        }
    }
}

#[derive(Debug, Clone)]
pub struct ClientOptions {
    pub connect_timeout: Duration,
    pub request_timeout: Duration,
    pub pool_idle_timeout: Duration,
    pub max_idle_connections_per_host: usize,
    pub max_response_bytes: usize,
    pub retry: RetryPolicy,
    pub user_agent: String,
    pub telemetry: Arc<dyn TelemetryObserver>,
    /// Permit plaintext HTTP only inside an authenticated service mesh.
    /// Loopback HTTP remains available without this opt-in for local tests.
    pub trusted_mesh_http: bool,
}

impl Default for ClientOptions {
    fn default() -> Self {
        Self {
            connect_timeout: Duration::from_secs(5),
            request_timeout: Duration::from_secs(30),
            pool_idle_timeout: Duration::from_secs(90),
            max_idle_connections_per_host: 16,
            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
            retry: RetryPolicy::default(),
            user_agent: format!("agent-infra-sdk/{}", env!("CARGO_PKG_VERSION")),
            telemetry: Arc::new(NoopTelemetry),
            trusted_mesh_http: false,
        }
    }
}

/// Per-call reliability controls. Retries remain disabled unless the caller
/// explicitly proves the operation idempotent.
#[derive(Debug, Clone, Default)]
pub struct CallOptions {
    pub deadline: Option<Duration>,
    /// Stable, non-secret operation correlation propagated as `x-request-id`.
    /// It does not by itself enable retries or assert idempotency.
    pub request_id: Option<String>,
    pub idempotency_key: Option<String>,
    pub idempotent: bool,
    pub traceparent: Option<String>,
    pub(crate) caller_credential: Option<BearerCredential>,
}

impl CallOptions {
    pub fn deadline(mut self, deadline: Duration) -> Self {
        self.deadline = Some(deadline);
        self
    }

    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
        self.idempotency_key = Some(key.into());
        self.idempotent = true;
        self
    }

    pub fn request_id(mut self, value: impl Into<String>) -> Self {
        self.request_id = Some(value.into());
        self
    }

    pub fn idempotent(mut self, value: bool) -> Self {
        self.idempotent = value;
        self
    }

    pub fn traceparent(mut self, value: impl Into<String>) -> Self {
        self.traceparent = Some(value.into());
        self
    }

    #[cfg(feature = "gateway")]
    pub(crate) fn caller_credential(mut self, value: BearerCredential) -> Self {
        self.caller_credential = Some(value);
        self
    }
}

#[derive(Debug)]
pub enum InfraClientError {
    ClientBuild(reqwest::Error),
    InvalidOptions {
        message: String,
    },
    Credential {
        service: &'static str,
        source: CredentialError,
    },
    InvalidEndpoint {
        service: &'static str,
        base_url: String,
        message: String,
    },
    Request {
        service: &'static str,
        source: reqwest::Error,
    },
    DeadlineExceeded {
        service: &'static str,
    },
    Canceled {
        operation_id: String,
    },
    OperationTerminal {
        operation_id: String,
        state: &'static str,
        code: Option<String>,
    },
    HttpStatus {
        service: &'static str,
        status: u16,
        code: String,
        message: String,
        retryable: bool,
        request_id: Option<String>,
        retry_after: Option<Duration>,
    },
    ResponseTooLarge {
        service: &'static str,
        limit: usize,
    },
    Decode {
        service: &'static str,
        source: serde_json::Error,
    },
    Protocol {
        service: &'static str,
        message: String,
    },
}

impl InfraClientError {
    pub fn code(&self) -> &str {
        match self {
            Self::ClientBuild(_) => "CLIENT_BUILD",
            Self::InvalidOptions { .. } => "INVALID_OPTIONS",
            Self::Credential { source, .. } => source.code(),
            Self::InvalidEndpoint { .. } => "INVALID_ENDPOINT",
            Self::Request { source, .. } if source.is_timeout() => "TIMEOUT",
            Self::Request { .. } => "TRANSPORT",
            Self::DeadlineExceeded { .. } => "TIMEOUT",
            Self::Canceled { .. } => "CANCELED",
            Self::OperationTerminal {
                code: Some(code), ..
            } => code,
            Self::OperationTerminal { .. } => "OPERATION_TERMINAL",
            Self::HttpStatus { code, .. } => code,
            Self::ResponseTooLarge { .. } => "RESPONSE_TOO_LARGE",
            Self::Decode { .. } => "DECODE",
            Self::Protocol { .. } => "PROTOCOL",
        }
    }

    pub fn retryable(&self) -> bool {
        match self {
            Self::Request { source, .. } => source.is_connect() || source.is_timeout(),
            Self::DeadlineExceeded { .. } => false,
            Self::Canceled { .. } | Self::OperationTerminal { .. } => false,
            Self::HttpStatus { retryable, .. } => *retryable,
            _ => false,
        }
    }

    pub fn request_id(&self) -> Option<&str> {
        match self {
            Self::HttpStatus { request_id, .. } => request_id.as_deref(),
            _ => None,
        }
    }
}

impl fmt::Display for InfraClientError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::ClientBuild(error) => {
                write!(formatter, "failed to build Infra HTTP client: {error}")
            }
            Self::InvalidOptions { message } => {
                write!(formatter, "invalid Infra client options: {message}")
            }
            Self::Credential { service, .. } => {
                write!(formatter, "failed to acquire {service} credential")
            }
            Self::InvalidEndpoint {
                service,
                base_url,
                message,
            } => {
                write!(
                    formatter,
                    "invalid {service} endpoint '{base_url}': {message}"
                )
            }
            Self::Request { service, source } => {
                write!(formatter, "{service} request failed: {source}")
            }
            Self::DeadlineExceeded { service } => {
                write!(formatter, "{service} request deadline elapsed")
            }
            Self::Canceled { operation_id } => {
                write!(formatter, "operation {operation_id} was canceled")
            }
            Self::OperationTerminal {
                operation_id,
                state,
                code,
            } => {
                write!(formatter, "operation {operation_id} ended in state {state}")?;
                if let Some(code) = code {
                    write!(formatter, " ({code})")?;
                }
                Ok(())
            }
            Self::HttpStatus {
                service,
                status,
                code,
                message,
                request_id,
                ..
            } => {
                write!(
                    formatter,
                    "{service} returned HTTP {status} ({code}): {message}"
                )?;
                if let Some(request_id) = request_id {
                    write!(formatter, " [request-id: {request_id}]")?;
                }
                Ok(())
            }
            Self::ResponseTooLarge { service, limit } => write!(
                formatter,
                "{service} response exceeded the configured {limit}-byte limit"
            ),
            Self::Decode { service, source } => write!(
                formatter,
                "{service} returned an invalid response: {source}"
            ),
            Self::Protocol { service, message } => write!(
                formatter,
                "{service} returned a rejected response: {message}"
            ),
        }
    }
}

impl Error for InfraClientError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            Self::ClientBuild(error) => Some(error),
            Self::Credential { source, .. } => Some(source),
            Self::Request { source, .. } => Some(source),
            Self::Decode { source, .. } => Some(source),
            _ => None,
        }
    }
}

#[derive(Clone)]
pub(crate) struct HttpTransport {
    http: Client,
    service: &'static str,
    base_url: Arc<str>,
    bearer_token: Option<Arc<str>>,
    credentials: Option<Arc<dyn CredentialsProvider>>,
    credential_audience: Arc<str>,
    options: ClientOptions,
}

impl fmt::Debug for HttpTransport {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("HttpTransport")
            .field("service", &self.service)
            .field("base_url", &redacted_endpoint(&self.base_url))
            .field("bearer_token_configured", &self.bearer_token.is_some())
            .field(
                "credentials_provider_configured",
                &self.credentials.is_some(),
            )
            .field("max_response_bytes", &self.options.max_response_bytes)
            .finish_non_exhaustive()
    }
}

#[allow(dead_code)] // Transport verbs are selected independently by feature-gated facades.
impl HttpTransport {
    pub(crate) fn new_with_options(
        http: Client,
        service: &'static str,
        endpoint: ServiceEndpoint,
        options: ClientOptions,
    ) -> Self {
        Self {
            http,
            service,
            base_url: endpoint.base_url.into(),
            bearer_token: endpoint.bearer_token.map(Into::into),
            credentials: endpoint.credentials,
            credential_audience: endpoint
                .credential_audience
                .unwrap_or_else(|| service.to_string())
                .into(),
            options,
        }
    }

    pub(crate) async fn get_json<Response>(&self, path: &str) -> Result<Response, InfraClientError>
    where
        Response: DeserializeOwned,
    {
        self.execute_json(
            Method::GET,
            path,
            None,
            CallOptions::default().idempotent(true),
        )
        .await
    }

    pub(crate) async fn get_json_with_options<Response>(
        &self,
        path: &str,
        options: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Response: DeserializeOwned,
    {
        self.execute_json(Method::GET, path, None, options.idempotent(true))
            .await
    }

    pub(crate) async fn post_json<Request, Response>(
        &self,
        path: &str,
        request: &Request,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        self.post_json_with_options(path, request, CallOptions::default())
            .await
    }

    pub(crate) async fn post_json_idempotent<Request, Response>(
        &self,
        path: &str,
        request: &Request,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        self.post_json_with_options(path, request, CallOptions::default().idempotent(true))
            .await
    }

    pub(crate) async fn post_json_with_options<Request, Response>(
        &self,
        path: &str,
        request: &Request,
        options: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        let body = serde_json::to_vec(request).map_err(|error| InfraClientError::Protocol {
            service: self.service,
            message: format!("failed to encode request: {error}"),
        })?;
        self.execute_json(Method::POST, path, Some(body), options)
            .await
    }

    pub(crate) async fn put_json_with_options<Request, Response>(
        &self,
        path: &str,
        request: &Request,
        options: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        let body = serde_json::to_vec(request).map_err(|error| InfraClientError::Protocol {
            service: self.service,
            message: format!("failed to encode request: {error}"),
        })?;
        self.execute_json(Method::PUT, path, Some(body), options)
            .await
    }

    pub(crate) async fn patch_json_with_options<Request, Response>(
        &self,
        path: &str,
        request: &Request,
        options: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        let body = serde_json::to_vec(request).map_err(|error| InfraClientError::Protocol {
            service: self.service,
            message: format!("failed to encode request: {error}"),
        })?;
        self.execute_json(Method::PATCH, path, Some(body), options)
            .await
    }

    pub(crate) async fn delete_json_with_options<Request, Response>(
        &self,
        path: &str,
        request: &Request,
        options: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Request: Serialize + ?Sized,
        Response: DeserializeOwned,
    {
        let body = serde_json::to_vec(request).map_err(|error| InfraClientError::Protocol {
            service: self.service,
            message: format!("failed to encode request: {error}"),
        })?;
        self.execute_json(Method::DELETE, path, Some(body), options)
            .await
    }

    fn url(&self, path: &str) -> Result<Url, InfraClientError> {
        let mut base =
            Url::parse(&self.base_url).map_err(|error| InfraClientError::InvalidEndpoint {
                service: self.service,
                base_url: redacted_endpoint(&self.base_url),
                message: error.to_string(),
            })?;
        if base.host_str().is_none() || !secure_transport(&base, self.options.trusted_mesh_http) {
            return Err(InfraClientError::InvalidEndpoint {
                service: self.service,
                base_url: redacted_endpoint(&self.base_url),
                message: "expected HTTPS, loopback HTTP, or explicitly trusted mesh HTTP".into(),
            });
        }
        if !base.username().is_empty() || base.password().is_some() {
            return Err(InfraClientError::InvalidEndpoint {
                service: self.service,
                base_url: redacted_endpoint(&self.base_url),
                message: "base URL must not contain embedded credentials".into(),
            });
        }
        if base.query().is_some() || base.fragment().is_some() {
            return Err(InfraClientError::InvalidEndpoint {
                service: self.service,
                base_url: redacted_endpoint(&self.base_url),
                message: "base URL must not contain a query or fragment".into(),
            });
        }
        let normalized_path = format!("{}/", base.path().trim_end_matches('/'));
        base.set_path(&normalized_path);
        base.join(path.trim_start_matches('/'))
            .map_err(|error| InfraClientError::InvalidEndpoint {
                service: self.service,
                base_url: redacted_endpoint(&self.base_url),
                message: error.to_string(),
            })
    }

    async fn execute_json<Response>(
        &self,
        method: Method,
        path: &str,
        body: Option<Vec<u8>>,
        call: CallOptions,
    ) -> Result<Response, InfraClientError>
    where
        Response: DeserializeOwned,
    {
        let url = self.url(path)?;
        let started = Instant::now();
        self.emit(TelemetryEvent {
            service: self.service,
            phase: TelemetryPhase::Start,
            attempt: 0,
            status: None,
            elapsed_micros: 0,
            response_bytes: None,
            retry_delay_millis: None,
            outcome: "started",
        });
        let traceparent = match call.traceparent.as_deref() {
            Some(value) if valid_traceparent(value) => value.to_string(),
            Some(_) => {
                return Err(InfraClientError::InvalidOptions {
                    message: "traceparent must be a valid W3C trace context value".into(),
                });
            }
            None => new_traceparent(),
        };
        // A per-call deadline may only tighten the client-wide budget. Retries
        // share this one wall-clock budget instead of resetting it per attempt.
        let deadline = call
            .deadline
            .map(|value| value.min(self.options.request_timeout))
            .unwrap_or(self.options.request_timeout);
        let attempts = if call.idempotent {
            self.options.retry.max_attempts.max(1)
        } else {
            1
        };
        let dynamic_credential = match &self.credentials {
            Some(provider) => {
                let remaining = deadline.saturating_sub(started.elapsed());
                if remaining.is_zero() {
                    self.emit_result(started, 0, None, None, "deadline");
                    return Err(InfraClientError::DeadlineExceeded {
                        service: self.service,
                    });
                }
                Some(
                    tokio::time::timeout(remaining, provider.credential(&self.credential_audience))
                        .await
                        .map_err(|_| {
                            self.emit_result(started, 0, None, None, "deadline");
                            InfraClientError::DeadlineExceeded {
                                service: self.service,
                            }
                        })?
                        .map_err(|source| InfraClientError::Credential {
                            service: self.service,
                            source,
                        })?,
                )
            }
            None => None,
        };
        // Serialize once and keep a reusable request template. Reqwest clones
        // its replayable byte body by reference, avoiding an O(payload) copy
        // for each retry attempt.
        let mut template = self
            .http
            .request(method, url)
            .header("user-agent", &self.options.user_agent)
            .header("traceparent", traceparent);
        if let Some(request_id) = &call.request_id {
            template = template.header("x-request-id", request_id);
        }
        if let Some(caller) = &call.caller_credential {
            template = template.header(
                infra_api_gateway_contract::CALLER_AUTHORIZATION_HEADER,
                format!("Bearer {}", caller.expose()),
            );
        }
        if let Some(credential) = &dynamic_credential {
            template = template.bearer_auth(credential.expose());
        } else if let Some(token) = &self.bearer_token {
            template = template.bearer_auth(token.as_ref());
        }
        if let Some(key) = &call.idempotency_key {
            template = template.header("Idempotency-Key", key);
        }
        if let Some(body) = body {
            template = template
                .header("content-type", "application/json")
                .body(body);
        }
        let template = template
            .build()
            .map_err(|source| InfraClientError::Request {
                service: self.service,
                source,
            })?;

        for attempt in 0..attempts {
            let remaining = deadline.saturating_sub(started.elapsed());
            if remaining == Duration::ZERO {
                self.emit_result(started, attempt + 1, None, None, "deadline");
                return Err(InfraClientError::DeadlineExceeded {
                    service: self.service,
                });
            }
            let mut request = template
                .try_clone()
                .ok_or_else(|| InfraClientError::Protocol {
                    service: self.service,
                    message: "request body cannot be replayed".into(),
                })?;
            let timeout = remaining;
            *request.timeout_mut() = Some(timeout);
            self.emit(TelemetryEvent {
                service: self.service,
                phase: TelemetryPhase::Attempt,
                attempt: attempt + 1,
                status: None,
                elapsed_micros: elapsed_micros(started),
                response_bytes: None,
                retry_delay_millis: None,
                outcome: "attempt",
            });

            let response = match self.http.execute(request).await {
                Ok(response) => response,
                Err(source) => {
                    let retryable = source.is_connect() || source.is_timeout();
                    if retryable && attempt + 1 < attempts {
                        let delay = self.retry_delay(attempt, None);
                        self.emit_retry(started, attempt + 1, None, delay);
                        self.sleep_before_retry(delay, started, Some(deadline))
                            .await?;
                        continue;
                    }
                    self.emit_result(
                        started,
                        attempt + 1,
                        None,
                        None,
                        if source.is_timeout() {
                            "timeout"
                        } else {
                            "transport"
                        },
                    );
                    return Err(InfraClientError::Request {
                        service: self.service,
                        source,
                    });
                }
            };
            let status = response.status();
            let request_id = response
                .headers()
                .get("x-request-id")
                .and_then(|value| value.to_str().ok())
                .map(ToOwned::to_owned);
            let retry_after = parse_retry_after(response.headers().get("retry-after"));
            let response_body = match crate::transport_body::read_body(
                response,
                self.options.max_response_bytes,
                self.service,
            )
            .await
            {
                Ok(body) => body,
                Err(error) => {
                    let outcome = match &error {
                        InfraClientError::ResponseTooLarge { .. } => "response_too_large",
                        InfraClientError::Request { .. } => "transport",
                        _ => "response_error",
                    };
                    self.emit_result(started, attempt + 1, Some(status.as_u16()), None, outcome);
                    return Err(error);
                }
            };
            if status.is_success() {
                // HTTP 204 and accepted command endpoints commonly have no
                // representation. Treat an empty successful body as JSON null
                // so callers can explicitly request `()` or `Option<T>`.
                let response_body = if response_body.is_empty() {
                    b"null".as_slice()
                } else {
                    response_body.as_slice()
                };
                let decoded = serde_json::from_slice(response_body).map_err(|source| {
                    InfraClientError::Decode {
                        service: self.service,
                        source,
                    }
                });
                self.emit_result(
                    started,
                    attempt + 1,
                    Some(status.as_u16()),
                    Some(response_body.len()),
                    if decoded.is_ok() { "success" } else { "decode" },
                );
                return decoded;
            }

            let envelope = error_envelope(&response_body);
            let status_retryable = matches!(status.as_u16(), 408 | 429 | 502 | 503 | 504);
            let retryable = envelope.retryable.unwrap_or(status_retryable);
            if status_retryable && envelope.retryable != Some(false) && attempt + 1 < attempts {
                let delay = self.retry_delay(attempt, retry_after);
                self.emit_retry(started, attempt + 1, Some(status.as_u16()), delay);
                self.sleep_before_retry(delay, started, Some(deadline))
                    .await?;
                continue;
            }
            self.emit_result(
                started,
                attempt + 1,
                Some(status.as_u16()),
                Some(response_body.len()),
                "http",
            );
            return Err(InfraClientError::HttpStatus {
                service: self.service,
                status: status.as_u16(),
                code: envelope
                    .code
                    .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())),
                message: envelope.message,
                retryable,
                request_id: envelope.request_id.or(request_id),
                retry_after,
            });
        }
        unreachable!("attempt count is always at least one")
    }

    fn retry_delay(&self, attempt: usize, retry_after: Option<Duration>) -> Duration {
        let multiplier = 1_u32
            .checked_shl(attempt.min(16) as u32)
            .unwrap_or(u32::MAX);
        let exponential = self.options.retry.base_delay.saturating_mul(multiplier);
        let jitter_bound =
            u64::try_from(self.options.retry.base_delay.as_millis().max(1)).unwrap_or(u64::MAX);
        let entropy = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .subsec_nanos() as u64
            ^ RETRY_JITTER_SEQUENCE
                .fetch_add(1, Ordering::Relaxed)
                .wrapping_mul(0x9e37_79b9_7f4a_7c15);
        let jitter = Duration::from_millis(entropy % jitter_bound);
        retry_after.unwrap_or_else(|| {
            exponential
                .saturating_add(jitter)
                .min(self.options.retry.max_delay)
        })
    }

    async fn sleep_before_retry(
        &self,
        delay: Duration,
        started: Instant,
        deadline: Option<Duration>,
    ) -> Result<(), InfraClientError> {
        if deadline.is_some_and(|deadline| started.elapsed().saturating_add(delay) >= deadline) {
            self.emit_result(started, 0, None, None, "deadline");
            return Err(InfraClientError::DeadlineExceeded {
                service: self.service,
            });
        }
        tokio::time::sleep(delay).await;
        Ok(())
    }

    fn emit_retry(&self, started: Instant, attempt: usize, status: Option<u16>, delay: Duration) {
        self.emit(TelemetryEvent {
            service: self.service,
            phase: TelemetryPhase::Retry,
            attempt,
            status,
            elapsed_micros: elapsed_micros(started),
            response_bytes: None,
            retry_delay_millis: Some(u64::try_from(delay.as_millis()).unwrap_or(u64::MAX)),
            outcome: "retry",
        });
    }

    fn emit_result(
        &self,
        started: Instant,
        attempt: usize,
        status: Option<u16>,
        response_bytes: Option<usize>,
        outcome: &'static str,
    ) {
        self.emit(TelemetryEvent {
            service: self.service,
            phase: TelemetryPhase::Result,
            attempt,
            status,
            elapsed_micros: elapsed_micros(started),
            response_bytes,
            retry_delay_millis: None,
            outcome,
        });
    }

    fn emit(&self, event: TelemetryEvent) {
        let observer = &self.options.telemetry;
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| observer.observe(event)));
    }
}

fn error_envelope(body: &[u8]) -> crate::transport_support::ParsedError {
    parse_error_envelope(body, MAX_ERROR_MESSAGE_BYTES)
}

#[cfg(test)]
#[path = "transport_tests.rs"]
mod tests;