lodestone_client/

transport.rs

use std::collections::VecDeque;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use lodestone_proto::{MAX_PUBLIC_SESSION_BYTES, PUBLIC_SESSION_FORMAT_VERSION};
use lodestone_wire::{
    Action, Command, CoreBinding, DiscoverRequest, DiscoverResult, ErrorBody, ErrorDisposition, GetBindingRequest,
    GetBindingResult, Identity, LODESTONE_API_ALPN, LODESTONE_API_COMMAND_CONTRACTS, LODESTONE_API_COMMAND_OUTCOMES,
    LODESTONE_API_COMMAND_SEMANTICS, LODESTONE_API_FORMAT_VERSION, LODESTONE_API_RESPONSE_META_FORMAT_VERSION,
    LODESTONE_API_SCHEMA_ID, LODESTONE_API_SPEC_SHA256, LODESTONE_API_STREAM_CONTRACTS, RawJson, RecoveryAction,
    RequestBody, RequestEnvelope, RequestMeta, ResponseEnvelope, ResultBody, U64, WorkloadClass, command_for_result,
    validate_error_body,
};
use rustls::pki_types::{CertificateDer, ServerName};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::{Mutex as AsyncMutex, Notify, Semaphore};
use tokio_rustls::TlsConnector;
use tokio_rustls::client::TlsStream;

use crate::config::{
    CellTransportConfig, CellTransportIdentity, ClientConfigError, CollectionSessionConfig, PoolConfig, parse_identity,
};
use crate::credential::{CredentialMaterial, CredentialProviderError};
use crate::generated::{
    GeneratedCollectionIndexRef, MAX_NATIVE_REQUEST_BYTES, MAX_NATIVE_RESPONSE_BYTES, request_collection_index_ref,
    request_command,
};

const MAX_REQUEST_ID_BYTES: usize = 128;
const MAX_CALL_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(30);
pub const MAX_VERIFIED_SESSION_METADATA: usize = 4_096;

#[derive(Clone)]
pub struct CallCancellation {
    inner: Arc<CallCancellationInner>,
}

struct CallCancellationInner {
    cancelled: AtomicBool,
    notify: Notify,
}

impl CallCancellation {
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: Arc::new(CallCancellationInner {
                cancelled: AtomicBool::new(false),
                notify: Notify::new(),
            }),
        }
    }

    pub fn cancel(&self) {
        if !self.inner.cancelled.swap(true, Ordering::AcqRel) {
            self.inner.notify.notify_waiters();
        }
    }

    #[must_use]
    pub fn is_cancelled(&self) -> bool {
        self.inner.cancelled.load(Ordering::Acquire)
    }

    /// Waits until this shared call scope is cancelled.
    pub async fn cancelled(&self) {
        loop {
            let notified = self.inner.notify.notified();
            if self.is_cancelled() {
                return;
            }
            notified.await;
        }
    }
}

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

#[derive(Clone)]
pub struct CallOptions {
    deadline: CallDeadline,
    cancellation: CallCancellation,
    index_name: Option<String>,
    workload_class: Option<WorkloadClass>,
}

#[derive(Clone, Copy)]
struct CallDeadline {
    absolute_unix_millis: Option<u64>,
    local: tokio::time::Instant,
}

impl CallOptions {
    /// Creates a call scope whose absolute deadline is anchored now.
    ///
    /// Cloning these options preserves that original deadline; it never turns
    /// the supplied duration into a new per-request timeout.
    pub fn new(timeout: Duration) -> Result<Self, ClientError> {
        Ok(Self {
            deadline: CallDeadline::after(timeout)?,
            cancellation: CallCancellation::new(),
            index_name: None,
            workload_class: None,
        })
    }

    /// Creates a call scope from an existing absolute Unix-millisecond
    /// deadline. The deadline must be in the future and no more than the
    /// client maximum call budget away.
    pub fn from_absolute_deadline_unix_millis(absolute_unix_millis: u64) -> Result<Self, ClientError> {
        Ok(Self {
            deadline: CallDeadline::from_absolute_unix_millis(absolute_unix_millis)?,
            cancellation: CallCancellation::new(),
            index_name: None,
            workload_class: None,
        })
    }

    #[must_use]
    pub fn with_cancellation(mut self, cancellation: CallCancellation) -> Self {
        self.cancellation = cancellation;
        self
    }

    pub fn with_index_name(mut self, index_name: String) -> Result<Self, ClientError> {
        validate_index_name(&index_name)?;
        self.index_name = Some(index_name);
        Ok(self)
    }

    /// Selects a workload class for this call. The server still requires the
    /// authenticated role to grant the class and checks action compatibility.
    #[must_use]
    pub fn with_workload_class(mut self, workload_class: WorkloadClass) -> Self {
        self.workload_class = Some(workload_class);
        self
    }

    #[must_use]
    pub const fn cancellation(&self) -> &CallCancellation {
        &self.cancellation
    }

    /// Returns the exact deadline placed in every request envelope in this
    /// call scope.
    pub fn absolute_deadline_unix_millis(&self) -> Result<u64, ClientError> {
        self.deadline.absolute_unix_millis.ok_or(ClientError::ClockUnavailable)
    }

    /// Returns the budget left in this call scope without resetting its
    /// absolute deadline.
    pub fn remaining(&self) -> Result<Duration, ClientError> {
        self.local_deadline()?
            .checked_duration_since(tokio::time::Instant::now())
            .ok_or(ClientError::Transport {
                kind: TransportErrorKind::Deadline,
                disposition: SubmissionDisposition::NotSubmitted,
            })
    }

    pub(crate) fn local_deadline(&self) -> Result<tokio::time::Instant, ClientError> {
        self.absolute_deadline_unix_millis().map(|_| self.deadline.local)
    }

    pub(crate) fn for_resolution(&self) -> Self {
        Self {
            deadline: self.deadline,
            cancellation: CallCancellation::new(),
            index_name: self.index_name.clone(),
            workload_class: self.workload_class,
        }
    }
}

impl CallDeadline {
    fn after(timeout: Duration) -> Result<Self, ClientError> {
        let timeout = normalized_timeout(timeout)?;
        let local_now = tokio::time::Instant::now();
        let wall_deadline = SystemTime::now()
            .checked_add(timeout)
            .ok_or(ClientError::ClockUnavailable)?;
        let since_epoch = wall_deadline
            .duration_since(UNIX_EPOCH)
            .map_err(|_| ClientError::ClockUnavailable)?;
        Ok(Self {
            absolute_unix_millis: Some(duration_millis_ceil(since_epoch)?),
            local: local_now + timeout,
        })
    }

    fn from_absolute_unix_millis(absolute_unix_millis: u64) -> Result<Self, ClientError> {
        let local_now = tokio::time::Instant::now();
        let wall_deadline = UNIX_EPOCH
            .checked_add(Duration::from_millis(absolute_unix_millis))
            .ok_or(ClientError::InvalidCallOptions)?;
        let remaining = wall_deadline
            .duration_since(SystemTime::now())
            .map_err(|_| ClientError::InvalidCallOptions)?;
        if remaining.is_zero() || remaining > MAX_CALL_TIMEOUT {
            return Err(ClientError::InvalidCallOptions);
        }
        Ok(Self {
            absolute_unix_millis: Some(absolute_unix_millis),
            local: local_now + remaining,
        })
    }
}

impl Default for CallOptions {
    fn default() -> Self {
        match Self::new(DEFAULT_CALL_TIMEOUT) {
            Ok(options) => options,
            Err(_) => Self {
                deadline: CallDeadline {
                    absolute_unix_millis: None,
                    local: tokio::time::Instant::now(),
                },
                cancellation: CallCancellation::new(),
                index_name: None,
                workload_class: None,
            },
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProfileVerification {
    pub artifact_version: String,
    pub principal_id: Identity,
    pub collection_id: Identity,
    pub policy_revision: U64,
    pub cluster_id: Identity,
    pub configuration_generation: U64,
    pub binding_revision: U64,
    pub schema_id: String,
    pub schema_digest: lodestone_wire::Digest,
    pub native_endpoints: Vec<String>,
    pub selected_stream_endpoint: String,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CellTransportMetricsSnapshot {
    pub transport_count: usize,
    pub live_connections: usize,
    pub idle_connections: usize,
    pub live_collection_sessions: usize,
    pub session_cache_entries: usize,
    pub session_cache_hits: u64,
    pub session_cache_misses: u64,
    pub session_cache_evictions: u64,
    pub credential_fetches: u64,
    pub bulkhead_saturation: u64,
    pub isolation_failures: u64,
}

#[derive(Default)]
struct CellTransportMetrics {
    live_connections: AtomicUsize,
    idle_connections: AtomicUsize,
    live_collection_sessions: AtomicUsize,
    session_cache_entries: AtomicUsize,
    session_cache_hits: AtomicU64,
    session_cache_misses: AtomicU64,
    session_cache_evictions: AtomicU64,
    credential_fetches: AtomicU64,
    bulkhead_saturation: AtomicU64,
    isolation_failures: AtomicU64,
}

impl CellTransportMetrics {
    fn snapshot(&self) -> CellTransportMetricsSnapshot {
        CellTransportMetricsSnapshot {
            transport_count: 1,
            live_connections: self.live_connections.load(Ordering::Relaxed),
            idle_connections: self.idle_connections.load(Ordering::Relaxed),
            live_collection_sessions: self.live_collection_sessions.load(Ordering::Relaxed),
            session_cache_entries: self.session_cache_entries.load(Ordering::Relaxed),
            session_cache_hits: self.session_cache_hits.load(Ordering::Relaxed),
            session_cache_misses: self.session_cache_misses.load(Ordering::Relaxed),
            session_cache_evictions: self.session_cache_evictions.load(Ordering::Relaxed),
            credential_fetches: self.credential_fetches.load(Ordering::Relaxed),
            bulkhead_saturation: self.bulkhead_saturation.load(Ordering::Relaxed),
            isolation_failures: self.isolation_failures.load(Ordering::Relaxed),
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct VerifiedSessionCacheKey {
    collection_id: String,
    principal_id: String,
    profile_generation: String,
    credential_id: String,
    credential_generation: String,
    secret_reference: String,
}

impl VerifiedSessionCacheKey {
    fn from_config(config: &CollectionSessionConfig) -> Self {
        let profile = config.profile.as_wire();
        Self {
            collection_id: profile.expected_collection_id.value.clone(),
            principal_id: profile.expected_principal_id.value.clone(),
            profile_generation: profile.profile_generation.value.clone(),
            credential_id: profile.credential_id.value.clone(),
            credential_generation: profile.credential_generation.value.clone(),
            secret_reference: profile.secret_reference.clone(),
        }
    }
}

struct VerifiedSessionMetadata {
    key: VerifiedSessionCacheKey,
    verification: ProfileVerification,
}

#[derive(Default)]
struct VerifiedSessionCache {
    entries: VecDeque<VerifiedSessionMetadata>,
}

impl VerifiedSessionCache {
    fn get(&mut self, key: &VerifiedSessionCacheKey, metrics: &CellTransportMetrics) -> Option<ProfileVerification> {
        let Some(position) = self.entries.iter().position(|entry| &entry.key == key) else {
            metrics.session_cache_misses.fetch_add(1, Ordering::Relaxed);
            return None;
        };
        let entry = self.entries.remove(position)?;
        let verification = entry.verification.clone();
        self.entries.push_back(entry);
        metrics.session_cache_hits.fetch_add(1, Ordering::Relaxed);
        Some(verification)
    }

    fn insert(
        &mut self,
        key: VerifiedSessionCacheKey,
        verification: ProfileVerification,
        metrics: &CellTransportMetrics,
    ) {
        if let Some(position) = self.entries.iter().position(|entry| entry.key == key) {
            let _ = self.entries.remove(position);
        } else if self.entries.len() == MAX_VERIFIED_SESSION_METADATA {
            let _ = self.entries.pop_front();
            metrics.session_cache_evictions.fetch_add(1, Ordering::Relaxed);
        }
        self.entries.push_back(VerifiedSessionMetadata { key, verification });
        metrics
            .session_cache_entries
            .store(self.entries.len(), Ordering::Relaxed);
    }
}

#[derive(Clone)]
pub struct CellTransport {
    inner: Arc<CellTransportInner>,
}

struct CellTransportInner {
    config: CellTransportConfig,
    tls: Arc<rustls::ClientConfig>,
    pool: AsyncMutex<Vec<NativeConnection>>,
    connection_slots: Arc<Semaphore>,
    next_endpoint: AtomicUsize,
    verified_sessions: Mutex<VerifiedSessionCache>,
    metrics: Arc<CellTransportMetrics>,
}

#[derive(Clone)]
pub struct Client {
    inner: Arc<ClientInner>,
}

pub type CollectionSession = Client;

struct ClientInner {
    transport: CellTransport,
    config: CollectionSessionConfig,
    verified: Mutex<Option<ProfileVerification>>,
}

impl Drop for ClientInner {
    fn drop(&mut self) {
        self.transport
            .inner
            .metrics
            .live_collection_sessions
            .fetch_sub(1, Ordering::Relaxed);
    }
}

struct NativeConnection {
    stream: TlsStream<TcpStream>,
    metrics: Arc<CellTransportMetrics>,
    idle_since: tokio::time::Instant,
}

impl Drop for NativeConnection {
    fn drop(&mut self) {
        self.metrics.live_connections.fetch_sub(1, Ordering::Relaxed);
    }
}

impl CellTransport {
    pub fn new(config: CellTransportConfig) -> Result<Self, ClientError> {
        let tls = build_tls_config(&config)?;
        let metrics = Arc::new(CellTransportMetrics::default());
        Ok(Self {
            inner: Arc::new(CellTransportInner {
                connection_slots: Arc::new(Semaphore::new(config.pool.maximum_connections)),
                config,
                tls: Arc::new(tls),
                pool: AsyncMutex::new(Vec::new()),
                next_endpoint: AtomicUsize::new(0),
                verified_sessions: Mutex::new(VerifiedSessionCache::default()),
                metrics,
            }),
        })
    }

    #[must_use]
    pub fn identity(&self) -> &CellTransportIdentity {
        self.inner.config.identity()
    }

    #[must_use]
    pub fn pool_config(&self) -> PoolConfig {
        self.inner.config.pool
    }

    #[must_use]
    pub fn metrics(&self) -> CellTransportMetricsSnapshot {
        self.inner.metrics.snapshot()
    }
}

impl Client {
    /// Connects, authenticates, discovers capabilities, and verifies the exact
    /// cluster, collection, schema, stream, and binding identities in the profile.
    pub async fn connect(
        transport: CellTransport,
        config: CollectionSessionConfig,
        options: CallOptions,
    ) -> Result<Self, ClientError> {
        let client = Self::new(transport, config)?;
        client.verify_connection_profile(options).await?;
        Ok(client)
    }

    pub fn new(transport: CellTransport, config: CollectionSessionConfig) -> Result<Self, ClientError> {
        if let Err(error) = config.validate_transport(&transport.inner.config) {
            transport
                .inner
                .metrics
                .isolation_failures
                .fetch_add(1, Ordering::Relaxed);
            return Err(error.into());
        }
        let cache_key = VerifiedSessionCacheKey::from_config(&config);
        let cached = transport
            .inner
            .verified_sessions
            .lock()
            .map_err(|_| ClientError::InternalStateUnavailable)?
            .get(&cache_key, &transport.inner.metrics);
        if let Some(verification) = &cached
            && validate_cached_verification(&config, verification).is_err()
        {
            transport
                .inner
                .metrics
                .isolation_failures
                .fetch_add(1, Ordering::Relaxed);
            return Err(ClientError::ProfileIdentityMismatch);
        }
        transport
            .inner
            .metrics
            .live_collection_sessions
            .fetch_add(1, Ordering::Relaxed);
        Ok(Self {
            inner: Arc::new(ClientInner {
                transport,
                config,
                // INVARIANT: shared metadata never authorizes a fresh session.
                // Every session performs its own authenticated discover/binding
                // exchange before mutation, even after an exact cache hit.
                verified: Mutex::new(None),
            }),
        })
    }

    #[must_use]
    pub fn connection_profile(&self) -> &crate::ValidatedConnectionProfile {
        &self.inner.config.profile
    }

    /// Returns a handle to the exact shared cell transport for creating
    /// additional isolated Collection sessions.
    #[must_use]
    pub fn cell_transport(&self) -> CellTransport {
        self.inner.transport.clone()
    }

    #[must_use]
    pub fn profile_verification(&self) -> Option<ProfileVerification> {
        self.inner.verified.lock().ok().and_then(|verified| verified.clone())
    }

    pub async fn verify_connection_profile(&self, options: CallOptions) -> Result<ProfileVerification, ClientError> {
        let discovered = self.discover(DiscoverRequest {}, options.clone()).await?;
        self.validate_discovery_observation(&discovered)?;
        let binding = self.get_binding(GetBindingRequest {}, options).await?;
        self.verify_authority_binding_observation(&discovered, &binding)
    }

    /// Validates one authenticated discovery observation against the selected
    /// profile without performing another network call.
    pub fn validate_discovery_observation(&self, discovered: &DiscoverResult) -> Result<(), ClientError> {
        validate_discovery(&self.inner.config, discovered)
    }

    /// Validates and installs one authenticated discovery/binding cut without
    /// performing another network call.
    pub fn verify_authority_binding_observation(
        &self,
        discovered: &DiscoverResult,
        binding: &GetBindingResult,
    ) -> Result<ProfileVerification, ClientError> {
        let verified = self.validate_authority_binding_observation(discovered, binding)?;
        self.install_profile_verification(verified)
    }

    /// Validates one authenticated discovery/binding cut without installing
    /// session authority. Connection diagnostics use this read-only probe so
    /// they cannot make a later mutation eligible or alter its disposition.
    pub fn validate_authority_binding_observation(
        &self,
        discovered: &DiscoverResult,
        binding: &GetBindingResult,
    ) -> Result<ProfileVerification, ClientError> {
        validate_discovery(&self.inner.config, discovered)?;
        validate_binding(&self.inner.config, &binding.binding, binding.derived, &binding.revision)?;
        Ok(ProfileVerification {
            artifact_version: discovered.artifact_version.clone(),
            principal_id: discovered.principal_id.clone(),
            collection_id: discovered.collection_id.clone(),
            policy_revision: discovered.policy_revision.clone(),
            cluster_id: binding.binding.cluster_id.clone(),
            configuration_generation: binding.binding.configuration_generation.clone(),
            binding_revision: binding.revision.clone(),
            schema_id: binding.binding.schema_id.clone(),
            schema_digest: binding.binding.schema_digest.clone(),
            native_endpoints: binding.binding.native_endpoints.clone(),
            selected_stream_endpoint: binding.binding.selected_stream_endpoint.clone(),
        })
    }

    fn install_profile_verification(&self, verified: ProfileVerification) -> Result<ProfileVerification, ClientError> {
        let mut slot = self
            .inner
            .verified
            .lock()
            .map_err(|_| ClientError::InternalStateUnavailable)?;
        *slot = Some(verified.clone());
        self.inner
            .transport
            .inner
            .verified_sessions
            .lock()
            .map_err(|_| ClientError::InternalStateUnavailable)?
            .insert(
                VerifiedSessionCacheKey::from_config(&self.inner.config),
                verified.clone(),
                &self.inner.transport.inner.metrics,
            );
        Ok(verified)
    }

    /// Dispatches one closed Lodestone API request body through the same validated
    /// transport and recovery boundary used by every generated typed method.
    ///
    /// The caller supplies no envelope, collection, credential, or arbitrary
    /// header. Those remain owned by the validated connection profile,
    /// credential provider, and call options.
    pub async fn call(&self, request: RequestBody, options: CallOptions) -> Result<ResultBody, ClientError> {
        let local_deadline = options.local_deadline()?;
        if tokio::time::Instant::now() >= local_deadline {
            return Err(ClientError::Transport {
                kind: TransportErrorKind::Deadline,
                disposition: SubmissionDisposition::NotSubmitted,
            });
        }
        let command = request_command(&request);
        let command_contract = LODESTONE_API_COMMAND_CONTRACTS
            .iter()
            .find(|contract| contract.command == command)
            .ok_or(ClientError::MissingCommandContract)?;
        let outcome_contract = LODESTONE_API_COMMAND_OUTCOMES
            .iter()
            .find(|contract| contract.command == command)
            .ok_or(ClientError::MissingCommandContract)?;
        let mutation = outcome_contract.possible_submission_states.contains(&"submitted");
        if mutation && self.profile_verification().is_none() {
            return Err(ClientError::ProfileNotVerified);
        }
        let index_name = resolve_index_name(&request, options.index_name.as_deref())?;
        let correlation = random_identity()?;
        let deadline = options.absolute_deadline_unix_millis()?;
        let request_id = format!("r-{}", lower_hex(&correlation));
        let workload_class = options
            .workload_class
            .unwrap_or_else(|| default_workload_class(command_contract.action));
        let envelope = RequestEnvelope {
            meta: RequestMeta {
                format_version: LODESTONE_API_FORMAT_VERSION,
                request_id: request_id.clone(),
                correlation_id: Identity {
                    value: lower_hex(&correlation),
                },
                absolute_deadline_unix_millis: U64 {
                    value: deadline.to_string(),
                },
                workload_class,
            },
            request,
        };
        let payload = canonical_request(&envelope)?;
        let started = Arc::new(AtomicBool::new(false));
        let maximum_attempts = usize::from(self.pool_config().maximum_read_retries).saturating_add(1);
        let mut attempt = 0_usize;
        loop {
            attempt = attempt.saturating_add(1);
            let result = tokio::select! {
                result = tokio::time::timeout_at(
                    local_deadline,
                    self.transport_attempt(
                        &payload,
                        &request_id,
                        command,
                        command_contract.action,
                        workload_class,
                        index_name.as_deref(),
                        correlation,
                        Arc::clone(&started),
                    ),
                ) => match result {
                    Ok(result) => result,
                    Err(_) => Err(ClientError::Transport {
                        kind: TransportErrorKind::Deadline,
                        disposition: started_disposition(&started),
                    }),
                },
                () = options.cancellation.cancelled() => Err(ClientError::Transport {
                    kind: TransportErrorKind::Cancelled,
                    disposition: started_disposition(&started),
                }),
            };
            match result {
                Err(ref error)
                    if attempt < maximum_attempts
                        && !options.cancellation.is_cancelled()
                        && error.automatic_retry_admitted() =>
                {
                    continue;
                }
                result => return result,
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn transport_attempt(
        &self,
        payload: &[u8],
        request_id: &str,
        command: Command,
        action: Action,
        workload_class: WorkloadClass,
        index_name: Option<&str>,
        correlation: [u8; 16],
        started: Arc<AtomicBool>,
    ) -> Result<ResultBody, ClientError> {
        // INVARIANT: Collection-local credential failures cannot consume or
        // discard a shared cell connection. Material remains request-local and
        // is fully encoded before this call enters the transport bulkhead.
        let credential = self.credential().await?;
        let native_credential = native_credential(&credential)?;
        let session = PublicSession {
            format_version: PUBLIC_SESSION_FORMAT_VERSION,
            api_key_credential: native_credential.as_slice(),
            collection_id: &self.inner.config.profile.as_wire().expected_collection_id.value,
            request_id,
            action: action_name(action),
            workload_class,
            index_name,
            request_correlation_id: correlation,
        };
        let session = SensitiveBuffer(serde_json::to_vec(&session).map_err(|_| ClientError::RequestCodec)?);
        if session.is_empty() || session.len() > MAX_PUBLIC_SESSION_BYTES {
            return Err(ClientError::RequestTooLarge);
        }
        let mut wire = SensitiveBuffer::new();
        wire.extend_u32(session.len())?;
        wire.extend(session.as_slice());
        wire.extend_u32(payload.len())?;
        wire.extend(payload);

        if self.inner.transport.inner.connection_slots.available_permits() == 0 {
            self.inner
                .transport
                .inner
                .metrics
                .bulkhead_saturation
                .fetch_add(1, Ordering::Relaxed);
        }
        let _permit = self
            .inner
            .transport
            .inner
            .connection_slots
            .clone()
            .acquire_owned()
            .await
            .map_err(|_| ClientError::Transport {
                kind: TransportErrorKind::PoolClosed,
                disposition: SubmissionDisposition::NotSubmitted,
            })?;
        let endpoint_index = self.inner.transport.inner.next_endpoint.fetch_add(1, Ordering::Relaxed)
            % self.inner.transport.inner.config.native_endpoints.len();
        let mut connection = {
            let mut pool = self.inner.transport.inner.pool.lock().await;
            // Every endpoint in the exact transport identity serves the same
            // cell. Reuse any idle connection so endpoint rotation cannot
            // create more live sockets than the transport connection bound.
            // The public gateway bounds how long it waits for the next
            // prelude, so an indefinitely idle pooled stream may already be
            // half-closed. Retire aged entries before a request can cross an
            // indeterminate write boundary on one of them.
            let now = tokio::time::Instant::now();
            let connection = loop {
                match pool.pop() {
                    Some(connection) if connection_idle_is_reusable(connection.idle_since, now) => {
                        break Some(connection);
                    }
                    Some(_) => {}
                    None => break None,
                }
            };
            self.inner
                .transport
                .inner
                .metrics
                .idle_connections
                .store(pool.len(), Ordering::Relaxed);
            connection
        };
        if connection.is_none() {
            connection = Some(self.connect_native(endpoint_index).await?);
        }
        let mut connection = connection.ok_or(ClientError::Transport {
            kind: TransportErrorKind::Connect,
            disposition: SubmissionDisposition::NotSubmitted,
        })?;
        started.store(true, Ordering::Release);
        if connection.stream.write_all(wire.as_slice()).await.is_err() || connection.stream.flush().await.is_err() {
            return Err(ClientError::Transport {
                kind: TransportErrorKind::Write,
                disposition: SubmissionDisposition::Indeterminate,
            });
        }
        let response_length = connection.stream.read_u32().await.map_err(|_| ClientError::Transport {
            kind: TransportErrorKind::Read,
            disposition: SubmissionDisposition::Indeterminate,
        })?;
        let response_length = usize::try_from(response_length).map_err(|_| ClientError::ResponseTooLarge)?;
        if response_length == 0 || response_length > MAX_NATIVE_RESPONSE_BYTES {
            return Err(ClientError::ResponseTooLarge);
        }
        let mut response = vec![0; response_length];
        connection
            .stream
            .read_exact(&mut response)
            .await
            .map_err(|_| ClientError::Transport {
                kind: TransportErrorKind::Read,
                disposition: SubmissionDisposition::Indeterminate,
            })?;
        let decoded: ResponseEnvelope = serde_json::from_slice(&response).map_err(|_| ClientError::ResponseCodec)?;
        if serde_json::to_vec(&decoded).map_err(|_| ClientError::ResponseCodec)? != response {
            return Err(ClientError::NonCanonicalResponse);
        }
        let result = match decoded {
            ResponseEnvelope::Success { meta, result } if response_meta_matches(&meta, request_id, &correlation) => {
                if command_for_result(&result) != command {
                    Err(ClientError::ResponseCommandMismatch)
                } else {
                    Ok(*result)
                }
            }
            ResponseEnvelope::Success { .. } => Err(ClientError::ResponseIdentityMismatch),
            ResponseEnvelope::Error { meta, error } => {
                if !response_meta_matches(&meta, request_id, &correlation) {
                    return Err(ClientError::ResponseIdentityMismatch);
                }
                validate_error_body(&error).map_err(|_| ClientError::InvalidErrorShape)?;
                let semantic = LODESTONE_API_COMMAND_SEMANTICS
                    .iter()
                    .find(|contract| contract.command == command)
                    .ok_or(ClientError::MissingCommandContract)?;
                if !semantic.admissible_errors.contains(&error.code) {
                    return Err(ClientError::InvalidErrorShape);
                }
                Err(ClientError::Server(error))
            }
        };
        // A success proves that the peer completed the request while keeping
        // the shared byte stream synchronized. A typed refusal does not prove
        // that the gateway kept the TLS session open: admission may close the
        // connection immediately after writing the refusal. Discard refusal
        // connections so a dependent follow-up (for example GetIndex
        // NotFound followed by PrepareCreateIndex) cannot reuse a half-closed
        // stream and consume its entire deadline waiting for a response.
        if response_keeps_connection(&result) {
            let mut pool = self.inner.transport.inner.pool.lock().await;
            if pool.len() < self.pool_config().maximum_idle_connections {
                connection.idle_since = tokio::time::Instant::now();
                pool.push(connection);
                self.inner
                    .transport
                    .inner
                    .metrics
                    .idle_connections
                    .store(pool.len(), Ordering::Relaxed);
            }
        }
        result
    }

    async fn connect_native(&self, endpoint_index: usize) -> Result<NativeConnection, ClientError> {
        let endpoint = self
            .inner
            .transport
            .inner
            .config
            .native_endpoints
            .get(endpoint_index)
            .ok_or(ClientError::InvalidEndpoint)?;
        let stream = tokio::time::timeout(
            self.pool_config().connect_timeout,
            TcpStream::connect((endpoint.host(), endpoint.port())),
        )
        .await
        .map_err(|_| ClientError::Transport {
            kind: TransportErrorKind::Connect,
            disposition: SubmissionDisposition::NotSubmitted,
        })?
        .map_err(|_| ClientError::Transport {
            kind: TransportErrorKind::Connect,
            disposition: SubmissionDisposition::NotSubmitted,
        })?;
        let server_name = ServerName::try_from(self.inner.transport.inner.config.identity.tls_server_name().to_owned())
            .map_err(|_| ClientError::Config(ClientConfigError::InvalidServerName))?;
        let connector = TlsConnector::from(Arc::clone(&self.inner.transport.inner.tls));
        let stream = tokio::time::timeout(
            self.pool_config().connect_timeout,
            connector.connect(server_name, stream),
        )
        .await
        .map_err(|_| ClientError::Transport {
            kind: TransportErrorKind::Tls,
            disposition: SubmissionDisposition::NotSubmitted,
        })?
        .map_err(|_| ClientError::Transport {
            kind: TransportErrorKind::Tls,
            disposition: SubmissionDisposition::NotSubmitted,
        })?;
        if stream.get_ref().1.alpn_protocol() != Some(LODESTONE_API_ALPN.as_bytes()) {
            return Err(ClientError::Transport {
                kind: TransportErrorKind::Tls,
                disposition: SubmissionDisposition::NotSubmitted,
            });
        }
        self.inner
            .transport
            .inner
            .metrics
            .live_connections
            .fetch_add(1, Ordering::Relaxed);
        Ok(NativeConnection {
            stream,
            metrics: Arc::clone(&self.inner.transport.inner.metrics),
            idle_since: tokio::time::Instant::now(),
        })
    }

    pub(crate) async fn credential(&self) -> Result<CredentialMaterial, ClientError> {
        let profile = self.inner.config.profile.as_wire();
        self.inner
            .transport
            .inner
            .metrics
            .credential_fetches
            .fetch_add(1, Ordering::Relaxed);
        let material = self
            .inner
            .config
            .credentials
            .provide(&profile.secret_reference)
            .await
            .map_err(ClientError::Credential)?;
        if material.api_key_id() != &profile.credential_id || material.generation() != &profile.credential_generation {
            return Err(ClientError::Credential(CredentialProviderError::InvalidMaterial));
        }
        Ok(material)
    }

    pub(crate) fn transport_config(&self) -> &CellTransportConfig {
        &self.inner.transport.inner.config
    }

    pub(crate) fn pool_config(&self) -> PoolConfig {
        self.inner.transport.pool_config()
    }
}

fn response_keeps_connection(result: &Result<ResultBody, ClientError>) -> bool {
    result.is_ok()
}

const MAX_REUSABLE_CONNECTION_IDLE: Duration = Duration::from_secs(1);

fn connection_idle_is_reusable(idle_since: tokio::time::Instant, now: tokio::time::Instant) -> bool {
    now.checked_duration_since(idle_since)
        .is_some_and(|idle| idle <= MAX_REUSABLE_CONNECTION_IDLE)
}

const fn default_workload_class(action: Action) -> WorkloadClass {
    match action {
        Action::DataRead | Action::SubscriptionRead => WorkloadClass::InteractiveSearch,
        Action::DataWrite | Action::Append | Action::ResolveAppend => WorkloadClass::StreamingIngest,
        Action::IdentityDiscovery
        | Action::IndexAdmin
        | Action::CatalogRead
        | Action::TaskRead
        | Action::DiagnosticsRead
        | Action::DiagnosticsCreate
        | Action::SecurityRead
        | Action::SecurityAdmin
        | Action::AuditRead
        | Action::QuotaRead
        | Action::QuotaAdmin
        | Action::BindingRead
        | Action::LifecycleAdmin
        | Action::AdmissionRead
        | Action::AdmissionAdmin
        | Action::UsageRead
        | Action::OperationRead
        | Action::OperationCancel
        | Action::CollectionExport
        | Action::CollectionRestore => WorkloadClass::BatchAnalytics,
    }
}

#[derive(serde::Serialize)]
struct PublicSession<'a> {
    format_version: u16,
    api_key_credential: &'a [u8],
    collection_id: &'a str,
    request_id: &'a str,
    action: &'static str,
    workload_class: WorkloadClass,
    #[serde(skip_serializing_if = "Option::is_none")]
    index_name: Option<&'a str>,
    request_correlation_id: [u8; 16],
}

fn canonical_request(envelope: &RequestEnvelope) -> Result<Vec<u8>, ClientError> {
    if envelope.meta.request_id.is_empty()
        || envelope.meta.request_id.len() > MAX_REQUEST_ID_BYTES
        || envelope.meta.request_id.bytes().any(|byte| byte.is_ascii_control())
    {
        return Err(ClientError::InvalidRequestIdentity);
    }
    let payload = serde_json::to_vec(envelope).map_err(|_| ClientError::RequestCodec)?;
    if payload.is_empty() || payload.len() > MAX_NATIVE_REQUEST_BYTES {
        return Err(ClientError::RequestTooLarge);
    }
    let decoded: RequestEnvelope = serde_json::from_slice(&payload).map_err(|_| ClientError::RequestCodec)?;
    if decoded != *envelope {
        return Err(ClientError::RequestCodec);
    }
    Ok(payload)
}

fn build_tls_config(config: &CellTransportConfig) -> Result<rustls::ClientConfig, ClientError> {
    let mut roots = rustls::RootCertStore::empty();
    for certificate in config.trust.certificates_der() {
        roots
            .add(CertificateDer::from(certificate.clone()))
            .map_err(|_| ClientError::Config(ClientConfigError::InvalidTrustMaterial))?;
    }
    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
    let mut tls = rustls::ClientConfig::builder_with_provider(provider)
        .with_protocol_versions(&[&rustls::version::TLS13])
        .map_err(|_| ClientError::Config(ClientConfigError::InvalidTrustMaterial))?
        .with_root_certificates(roots)
        .with_no_client_auth();
    tls.alpn_protocols = vec![LODESTONE_API_ALPN.as_bytes().to_vec()];
    Ok(tls)
}

fn validate_discovery(config: &CollectionSessionConfig, discovered: &DiscoverResult) -> Result<(), ClientError> {
    let profile = config.profile.as_wire();
    if discovered.collection_id != profile.expected_collection_id
        || discovered.principal_id != profile.expected_principal_id
        || discovered.schema_id != LODESTONE_API_SCHEMA_ID
        || discovered.schema_digest.value != LODESTONE_API_SPEC_SHA256
    {
        return Err(ClientError::ProfileIdentityMismatch);
    }
    let expected = LODESTONE_API_STREAM_CONTRACTS
        .first()
        .ok_or(ClientError::MissingCommandContract)?;
    if discovered.streams.len() != 1
        || discovered.streams.first().is_none_or(|stream| {
            stream.name != expected.name
                || stream.transport != expected.transport
                || stream.service != expected.service
                || stream.method != expected.method
                || stream.endpoint != profile.selected_stream_endpoint
                || stream.auth_metadata != expected.auth_metadata
                || stream.auth_scheme != expected.auth_scheme
                || stream.semantic_format_version != u32::from(expected.semantic_format_version)
                || stream.schema_digest.value != LODESTONE_API_SPEC_SHA256
        })
    {
        return Err(ClientError::ProfileStreamMismatch);
    }
    Ok(())
}

fn validate_binding(
    config: &CollectionSessionConfig,
    binding: &CoreBinding,
    derived: bool,
    revision: &U64,
) -> Result<(), ClientError> {
    let profile = config.profile.as_wire();
    if !derived
        || binding.cluster_id != profile.expected_cluster_id
        || binding.collection_id != profile.expected_collection_id
        || binding.native_endpoints != profile.native_endpoints
        || binding.selected_stream_endpoint != profile.selected_stream_endpoint
        || binding.schema_id != profile.schema_id
        || binding.schema_digest != profile.schema_digest
        || binding
            .configuration_generation
            .value
            .parse::<u64>()
            .ok()
            .is_none_or(|generation| generation == 0)
        || revision.value.parse::<u64>().ok().is_none_or(|revision| revision == 0)
    {
        return Err(ClientError::ProfileIdentityMismatch);
    }
    Ok(())
}

fn validate_cached_verification(
    config: &CollectionSessionConfig,
    verification: &ProfileVerification,
) -> Result<(), ClientError> {
    let profile = config.profile.as_wire();
    if verification.collection_id != profile.expected_collection_id
        || verification.principal_id != profile.expected_principal_id
        || verification.cluster_id != profile.expected_cluster_id
        || verification.schema_id != profile.schema_id
        || verification.schema_digest != profile.schema_digest
        || verification.native_endpoints != profile.native_endpoints
        || verification.selected_stream_endpoint != profile.selected_stream_endpoint
        || verification
            .configuration_generation
            .value
            .parse::<u64>()
            .ok()
            .is_none_or(|generation| generation == 0)
        || verification
            .binding_revision
            .value
            .parse::<u64>()
            .ok()
            .is_none_or(|revision| revision == 0)
    {
        return Err(ClientError::ProfileIdentityMismatch);
    }
    Ok(())
}

fn resolve_index_name(request: &RequestBody, supplied: Option<&str>) -> Result<Option<String>, ClientError> {
    let scope = request_collection_index_ref(request)?;
    let inferred = match &scope {
        GeneratedCollectionIndexRef::Exact(name) => Some(name.clone()),
        GeneratedCollectionIndexRef::None | GeneratedCollectionIndexRef::Caller => None,
    };
    if let Some(supplied) = supplied {
        validate_index_name(supplied)?;
        if inferred.as_deref().is_some_and(|inferred| inferred != supplied) {
            return Err(ClientError::CollectionIndexRefMismatch);
        }
    }
    let resolved = supplied.map(str::to_owned).or(inferred);
    if let Some(name) = &resolved {
        validate_index_name(name)?;
    }
    match scope {
        GeneratedCollectionIndexRef::Caller | GeneratedCollectionIndexRef::Exact(_) if resolved.is_none() => {
            return Err(ClientError::CollectionIndexRefRequired);
        }
        GeneratedCollectionIndexRef::None if resolved.is_some() => return Err(ClientError::CollectionIndexRefMismatch),
        GeneratedCollectionIndexRef::None
        | GeneratedCollectionIndexRef::Caller
        | GeneratedCollectionIndexRef::Exact(_) => {}
    }
    Ok(resolved)
}

const fn action_name(action: Action) -> &'static str {
    match action {
        Action::IdentityDiscovery => "identity_discovery",
        Action::DataRead => "data_read",
        Action::DataWrite => "data_write",
        Action::Append => "append",
        Action::ResolveAppend => "resolve_append",
        Action::IndexAdmin => "index_admin",
        Action::CatalogRead => "catalog_read",
        Action::TaskRead => "task_read",
        Action::SubscriptionRead => "subscription_read",
        Action::DiagnosticsRead => "diagnostics_read",
        Action::DiagnosticsCreate => "diagnostics_create",
        Action::SecurityRead => "security_read",
        Action::SecurityAdmin => "security_admin",
        Action::AuditRead => "audit_read",
        Action::QuotaRead => "quota_read",
        Action::QuotaAdmin => "quota_admin",
        Action::BindingRead => "binding_read",
        Action::LifecycleAdmin => "lifecycle_admin",
        Action::AdmissionRead => "admission_read",
        Action::AdmissionAdmin => "admission_admin",
        Action::UsageRead => "usage_read",
        Action::OperationRead => "operation_read",
        Action::OperationCancel => "operation_cancel",
        Action::CollectionExport => "collection_export",
        Action::CollectionRestore => "collection_restore",
    }
}

fn native_credential(material: &CredentialMaterial) -> Result<SensitiveBuffer, ClientError> {
    let identifier = parse_identity(&material.api_key_id().value)
        .ok_or(ClientError::Credential(CredentialProviderError::InvalidMaterial))?;
    let mut output = SensitiveBuffer::new();
    output.extend(identifier.to_string().as_bytes());
    output.extend(b":");
    output.extend(&base64_url_no_pad(material.secret().expose()));
    Ok(output)
}

pub(crate) fn authorization_value(material: &CredentialMaterial) -> Result<SensitiveBuffer, ClientError> {
    let native = native_credential(material)?;
    let mut output = SensitiveBuffer::new();
    output.extend(b"ApiKey ");
    output.extend(&base64_standard(native.as_slice()));
    Ok(output)
}

fn base64_url_no_pad(bytes: &[u8]) -> Vec<u8> {
    base64_encode(
        bytes,
        b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_",
        false,
    )
}

fn base64_standard(bytes: &[u8]) -> Vec<u8> {
    base64_encode(
        bytes,
        b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/",
        true,
    )
}

/// Converts exact UTF-8 text into the authoritative Lodestone API raw JSON wire
/// leaf without parsing or normalizing the text.
pub fn raw_json_text(text: &str) -> Result<RawJson, ClientError> {
    let bytes_base64 = String::from_utf8(base64_standard(text.as_bytes())).map_err(|_| ClientError::RequestCodec)?;
    Ok(RawJson { bytes_base64 })
}

fn base64_encode(bytes: &[u8], alphabet: &[u8; 64], padded: bool) -> Vec<u8> {
    let mut output = Vec::with_capacity(bytes.len().div_ceil(3).saturating_mul(4));
    for chunk in bytes.chunks(3) {
        let first = chunk[0];
        let second = chunk.get(1).copied().unwrap_or(0);
        let third = chunk.get(2).copied().unwrap_or(0);
        output.push(alphabet[usize::from(first >> 2)]);
        output.push(alphabet[usize::from(((first & 0x03) << 4) | (second >> 4))]);
        if chunk.len() > 1 {
            output.push(alphabet[usize::from(((second & 0x0f) << 2) | (third >> 6))]);
        } else if padded {
            output.push(b'=');
        }
        if chunk.len() > 2 {
            output.push(alphabet[usize::from(third & 0x3f)]);
        } else if padded {
            output.push(b'=');
        }
    }
    output
}

fn random_identity() -> Result<[u8; 16], ClientError> {
    let mut output = [0; 16];
    getrandom::fill(&mut output).map_err(|_| ClientError::RandomUnavailable)?;
    if output == [0; 16] {
        return Err(ClientError::RandomUnavailable);
    }
    Ok(output)
}

fn normalized_timeout(timeout: Duration) -> Result<Duration, ClientError> {
    if timeout.is_zero() || timeout > MAX_CALL_TIMEOUT {
        return Err(ClientError::InvalidCallOptions);
    }
    let mut millis = timeout.as_millis();
    if !timeout.subsec_nanos().is_multiple_of(1_000_000) {
        millis = millis.checked_add(1).ok_or(ClientError::InvalidCallOptions)?;
    }
    let millis = u64::try_from(millis).map_err(|_| ClientError::InvalidCallOptions)?;
    let normalized = Duration::from_millis(millis);
    if normalized > MAX_CALL_TIMEOUT {
        return Err(ClientError::InvalidCallOptions);
    }
    Ok(normalized)
}

fn duration_millis_ceil(duration: Duration) -> Result<u64, ClientError> {
    let mut millis = duration.as_millis();
    if !duration.subsec_nanos().is_multiple_of(1_000_000) {
        millis = millis.checked_add(1).ok_or(ClientError::ClockUnavailable)?;
    }
    u64::try_from(millis).map_err(|_| ClientError::ClockUnavailable)
}

pub(crate) fn validate_index_name(value: &str) -> Result<(), ClientError> {
    if value.is_empty()
        || value.len() > 255
        || value
            .bytes()
            .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
    {
        return Err(ClientError::InvalidCollectionIndexRef);
    }
    Ok(())
}

fn lower_hex(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(bytes.len().saturating_mul(2));
    for byte in bytes {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    output
}

pub(crate) struct SensitiveBuffer(Vec<u8>);

impl SensitiveBuffer {
    fn new() -> Self {
        Self(Vec::new())
    }

    fn extend(&mut self, bytes: &[u8]) {
        self.0.extend_from_slice(bytes);
    }

    fn extend_u32(&mut self, value: usize) -> Result<(), ClientError> {
        let value = u32::try_from(value).map_err(|_| ClientError::RequestTooLarge)?;
        self.extend(&value.to_be_bytes());
        Ok(())
    }

    fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    fn len(&self) -> usize {
        self.0.len()
    }

    pub(crate) fn as_slice(&self) -> &[u8] {
        &self.0
    }
}

impl Drop for SensitiveBuffer {
    fn drop(&mut self) {
        self.0.fill(0);
        std::hint::black_box(&mut self.0);
    }
}

fn started_disposition(started: &AtomicBool) -> SubmissionDisposition {
    if started.load(Ordering::Acquire) {
        SubmissionDisposition::Indeterminate
    } else {
        SubmissionDisposition::NotSubmitted
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SubmissionDisposition {
    NotSubmitted,
    Submitted,
    Resolved,
    KnownRetired,
    Indeterminate,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransportErrorKind {
    PoolClosed,
    Connect,
    Tls,
    Write,
    Read,
    Deadline,
    Cancelled,
}

#[derive(Debug)]
pub enum ClientError {
    Config(ClientConfigError),
    Credential(CredentialProviderError),
    InvalidCallOptions,
    InvalidEndpoint,
    InvalidCollectionIndexRef,
    InvalidStateIndex,
    InvalidStateActionId,
    InvalidStateKey,
    InvalidStateBatch,
    UnexpectedStateResult,
    CollectionIndexRefRequired,
    CollectionIndexRefMismatch,
    ProfileNotVerified,
    ProfileIdentityMismatch,
    ProfileStreamMismatch,
    MissingCommandContract,
    RequestCodec,
    RequestTooLarge,
    InvalidRequestIdentity,
    ResponseCodec,
    ResponseTooLarge,
    NonCanonicalResponse,
    ResponseIdentityMismatch,
    ResponseCommandMismatch,
    InvalidErrorShape,
    RandomUnavailable,
    ClockUnavailable,
    InternalStateUnavailable,
    Transport {
        kind: TransportErrorKind,
        disposition: SubmissionDisposition,
    },
    Server(ErrorBody),
}

impl ClientError {
    #[must_use]
    pub fn submission_disposition(&self) -> SubmissionDisposition {
        match self {
            Self::Transport { disposition, .. } => *disposition,
            Self::Server(error) => server_error_disposition(error),
            Self::UnexpectedStateResult
            | Self::ResponseCodec
            | Self::ResponseTooLarge
            | Self::NonCanonicalResponse
            | Self::ResponseIdentityMismatch
            | Self::ResponseCommandMismatch
            | Self::InvalidErrorShape => SubmissionDisposition::Indeterminate,
            _ => SubmissionDisposition::NotSubmitted,
        }
    }

    /// Returns whether the current typed outcome permits an automatic replay.
    /// Indeterminate transport outcomes are always resolution-required; no
    /// command-level boolean may widen this decision.
    #[must_use]
    pub fn automatic_retry_admitted(&self) -> bool {
        matches!(
            self.retry_directive(),
            Some((
                ErrorDisposition::NotSubmitted,
                RecoveryAction::Retry | RecoveryAction::RetryAfter
            ))
        )
    }

    fn retry_directive(&self) -> Option<(ErrorDisposition, RecoveryAction)> {
        match self {
            Self::Server(error) => Some((error.disposition, error.recovery)),
            Self::Transport {
                kind: TransportErrorKind::Deadline | TransportErrorKind::Cancelled,
                ..
            } => Some((ErrorDisposition::NotSubmitted, RecoveryAction::Stop)),
            Self::Transport {
                disposition: SubmissionDisposition::NotSubmitted,
                ..
            } => Some((ErrorDisposition::NotSubmitted, RecoveryAction::Retry)),
            Self::Transport { .. } => Some((ErrorDisposition::Indeterminate, RecoveryAction::Resolve)),
            _ => None,
        }
    }

    pub(crate) fn is_deadline_exceeded(&self) -> bool {
        matches!(
            self,
            Self::Transport {
                kind: TransportErrorKind::Deadline,
                ..
            } | Self::Server(ErrorBody {
                code: lodestone_wire::ErrorCode::DeadlineExceeded,
                ..
            })
        )
    }
}

fn server_error_disposition(error: &ErrorBody) -> SubmissionDisposition {
    match error.disposition {
        lodestone_wire::ErrorDisposition::NotSubmitted => SubmissionDisposition::NotSubmitted,
        lodestone_wire::ErrorDisposition::Submitted => SubmissionDisposition::Submitted,
        lodestone_wire::ErrorDisposition::Resolved => SubmissionDisposition::Resolved,
        lodestone_wire::ErrorDisposition::KnownRetired => SubmissionDisposition::KnownRetired,
        lodestone_wire::ErrorDisposition::Indeterminate => SubmissionDisposition::Indeterminate,
    }
}

fn response_meta_matches(meta: &lodestone_wire::ResponseMeta, request_id: &str, correlation: &[u8; 16]) -> bool {
    meta.format_version == u32::from(LODESTONE_API_RESPONSE_META_FORMAT_VERSION)
        && meta.request_id == request_id
        && meta.correlation_id.value == lower_hex(correlation)
        && canonical_identity(&meta.diagnostic_id.value)
}

fn canonical_identity(value: &str) -> bool {
    value.len() == 32
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        && value.bytes().any(|byte| byte != b'0')
}

impl From<ClientConfigError> for ClientError {
    fn from(error: ClientConfigError) -> Self {
        Self::Config(error)
    }
}

impl Display for ClientError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Config(_) => "Core client configuration is invalid",
            Self::Credential(_) => "Core credential material is unavailable or mismatched",
            Self::InvalidCallOptions => "Core call options are invalid",
            Self::InvalidEndpoint => "Core endpoint is invalid",
            Self::InvalidCollectionIndexRef => "Core request index scope is invalid",
            Self::InvalidStateIndex => "Lodestone state index identity is invalid",
            Self::InvalidStateActionId => "Lodestone state action identity is invalid",
            Self::InvalidStateKey => "Lodestone state key must contain 1 to 254 bytes",
            Self::InvalidStateBatch => "Lodestone state key batch is empty or exceeds its bound",
            Self::UnexpectedStateResult => "Lodestone state result differs from the requested operation",
            Self::CollectionIndexRefRequired => "Core request requires an exact index scope",
            Self::CollectionIndexRefMismatch => "Core request and admission index scopes differ",
            Self::ProfileNotVerified => "Core mutation is refused until discovery and binding are verified",
            Self::ProfileIdentityMismatch => "Core discovery or binding differs from the connection profile",
            Self::ProfileStreamMismatch => "Core selected stream differs from the connection profile",
            Self::MissingCommandContract => "generated Core command metadata is incomplete",
            Self::RequestCodec => "Core request is not a canonical current Lodestone API value",
            Self::RequestTooLarge => "Core request exceeds its generated byte bound",
            Self::InvalidRequestIdentity => "Core request identity is invalid",
            Self::ResponseCodec => "Core response is malformed or unsupported",
            Self::ResponseTooLarge => "Core response exceeds its generated byte bound",
            Self::NonCanonicalResponse => "Core response encoding is noncanonical",
            Self::ResponseIdentityMismatch => "Core response changed the request identity or version",
            Self::ResponseCommandMismatch => "Core response returned another command result",
            Self::InvalidErrorShape => "Core error detail does not match its generated code",
            Self::RandomUnavailable => "secure request identity generation is unavailable",
            Self::ClockUnavailable => "absolute Core request deadline cannot be calculated",
            Self::InternalStateUnavailable => "Core client verification state is unavailable",
            Self::Transport { .. } => "Core transport failed",
            Self::Server(_) => "Core returned a typed request error",
        })
    }
}

impl Error for ClientError {}

#[cfg(test)]
mod tests {
    use std::error::Error;

    use lodestone_wire::{GetIndexRequest, Identity, OpenPitRequest, Resource, U64};

    use super::*;
    use crate::SecretBytes;

    #[test]
    fn response_failures_preserve_indeterminate_submission() {
        for error in [
            ClientError::UnexpectedStateResult,
            ClientError::ResponseCodec,
            ClientError::ResponseTooLarge,
            ClientError::NonCanonicalResponse,
            ClientError::ResponseIdentityMismatch,
            ClientError::ResponseCommandMismatch,
            ClientError::InvalidErrorShape,
        ] {
            assert_eq!(error.submission_disposition(), SubmissionDisposition::Indeterminate);
        }
        assert_eq!(
            ClientError::RequestCodec.submission_disposition(),
            SubmissionDisposition::NotSubmitted
        );
    }

    #[test]
    fn response_metadata_requires_exact_echo_and_nonzero_canonical_diagnostic() {
        let correlation = [0x12; 16];
        let mut meta = lodestone_wire::ResponseMeta {
            format_version: u32::from(LODESTONE_API_RESPONSE_META_FORMAT_VERSION),
            request_id: "request-1".to_owned(),
            correlation_id: Identity {
                value: lower_hex(&correlation),
            },
            diagnostic_id: Identity { value: "34".repeat(16) },
        };
        assert!(response_meta_matches(&meta, "request-1", &correlation));

        meta.correlation_id.value = "56".repeat(16);
        assert!(!response_meta_matches(&meta, "request-1", &correlation));
        meta.correlation_id.value = lower_hex(&correlation);

        meta.diagnostic_id.value = "00".repeat(16);
        assert!(!response_meta_matches(&meta, "request-1", &correlation));
        meta.diagnostic_id.value = "AB".repeat(16);
        assert!(!response_meta_matches(&meta, "request-1", &correlation));
        meta.diagnostic_id.value = "34".repeat(16);

        meta.request_id = "request-2".to_owned();
        assert!(!response_meta_matches(&meta, "request-1", &correlation));
        meta.request_id = "request-1".to_owned();
        meta.format_version = meta.format_version.saturating_add(1);
        assert!(!response_meta_matches(&meta, "request-1", &correlation));
    }

    fn cache_key(collection: u128, principal: u128, credential: u128) -> VerifiedSessionCacheKey {
        VerifiedSessionCacheKey {
            collection_id: format!("{collection:032x}"),
            principal_id: format!("{principal:032x}"),
            profile_generation: "1".to_owned(),
            credential_id: format!("{credential:032x}"),
            credential_generation: "1".to_owned(),
            secret_reference: format!("memory://session/{collection}/{principal}/{credential}"),
        }
    }

    fn verification(collection: u128, principal: u128) -> ProfileVerification {
        ProfileVerification {
            artifact_version: "development".to_owned(),
            principal_id: Identity {
                value: format!("{principal:032x}"),
            },
            collection_id: Identity {
                value: format!("{collection:032x}"),
            },
            policy_revision: U64 { value: "1".to_owned() },
            cluster_id: Identity {
                value: "00000000000000000000000000000001".to_owned(),
            },
            configuration_generation: U64 { value: "1".to_owned() },
            binding_revision: U64 { value: "1".to_owned() },
            schema_id: LODESTONE_API_SCHEMA_ID.to_owned(),
            schema_digest: lodestone_wire::Digest {
                value: LODESTONE_API_SPEC_SHA256.to_owned(),
            },
            native_endpoints: vec!["lodestone+tls://core.example:8443".to_owned()],
            selected_stream_endpoint: "https://core.example:9443".to_owned(),
        }
    }

    #[test]
    fn automatic_retry_uses_only_typed_disposition_and_recovery() {
        let transport = |disposition| ClientError::Transport {
            kind: TransportErrorKind::Read,
            disposition,
        };
        assert!(transport(SubmissionDisposition::NotSubmitted).automatic_retry_admitted());
        assert!(!transport(SubmissionDisposition::Submitted).automatic_retry_admitted());
        assert!(!transport(SubmissionDisposition::Indeterminate).automatic_retry_admitted());

        let server = |disposition, recovery| {
            ClientError::Server(ErrorBody {
                code: lodestone_wire::ErrorCode::TemporarilyUnavailable,
                message: "typed retry probe".to_owned(),
                disposition,
                recovery,
                details: lodestone_wire::ErrorDetail::TemporarilyUnavailable {
                    retry_after_millis: None,
                },
            })
        };
        assert!(server(ErrorDisposition::NotSubmitted, RecoveryAction::Retry).automatic_retry_admitted());
        assert!(server(ErrorDisposition::NotSubmitted, RecoveryAction::RetryAfter).automatic_retry_admitted());
        assert!(!server(ErrorDisposition::Indeterminate, RecoveryAction::Resolve).automatic_retry_admitted());
        assert!(!server(ErrorDisposition::NotSubmitted, RecoveryAction::Stop).automatic_retry_admitted());
    }

    #[test]
    fn one_deadline_survives_connect_state_and_resolution_clones() -> Result<(), Box<dyn Error>> {
        let options = CallOptions::new(Duration::from_secs(5))?;
        let absolute = options.absolute_deadline_unix_millis()?;
        let local = options.local_deadline()?;

        // These clones stand for discovery, binding, prepare, submit, and
        // resolution. None may derive a fresh duration from the current time.
        for stage in [
            options.clone(),
            options.clone(),
            options.clone(),
            options.clone(),
            options.for_resolution(),
        ] {
            assert_eq!(stage.absolute_deadline_unix_millis()?, absolute);
            assert_eq!(stage.local_deadline()?, local);
        }

        let resolution = options.for_resolution();
        options.cancellation().cancel();
        assert!(options.cancellation().is_cancelled());
        assert!(!resolution.cancellation().is_cancelled());
        assert_eq!(resolution.absolute_deadline_unix_millis()?, absolute);
        Ok(())
    }

    #[test]
    fn caller_supplied_absolute_deadline_is_preserved_exactly() -> Result<(), Box<dyn Error>> {
        let now = SystemTime::now().duration_since(UNIX_EPOCH)?;
        let now_millis = u64::try_from(now.as_millis())?;
        let absolute = now_millis.checked_add(5_000).ok_or("test deadline overflow")?;
        let options = CallOptions::from_absolute_deadline_unix_millis(absolute)?;
        assert_eq!(options.absolute_deadline_unix_millis()?, absolute);

        let too_far = now_millis
            .checked_add(u64::try_from(MAX_CALL_TIMEOUT.as_millis())?)
            .and_then(|value| value.checked_add(1_000))
            .ok_or("test deadline overflow")?;
        assert!(matches!(
            CallOptions::from_absolute_deadline_unix_millis(too_far),
            Err(ClientError::InvalidCallOptions)
        ));
        assert_eq!(normalized_timeout(Duration::from_nanos(1))?, Duration::from_millis(1));
        Ok(())
    }

    #[test]
    fn private_api_key_encodings_match_the_core_admission_contract() -> Result<(), Box<dyn Error>> {
        let material = CredentialMaterial::new(
            Identity {
                value: "00000000000000000000000000000001".to_owned(),
            },
            U64 { value: "1".to_owned() },
            SecretBytes::from_array([0; 32]),
        )?;
        assert_eq!(
            native_credential(&material)?.as_slice(),
            b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
        );
        assert_eq!(
            authorization_value(&material)?.as_slice(),
            b"ApiKey MTpBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFB"
        );
        Ok(())
    }

    #[test]
    fn typed_refusal_discards_the_connection_before_a_dependent_follow_up() {
        let refusal = Err(ClientError::Server(ErrorBody {
            code: lodestone_wire::ErrorCode::PermissionDenied,
            message: "denied".to_owned(),
            disposition: lodestone_wire::ErrorDisposition::NotSubmitted,
            recovery: lodestone_wire::RecoveryAction::Stop,
            details: lodestone_wire::ErrorDetail::PermissionDenied {
                action: None,
                workload_class: None,
            },
        }));
        assert!(!response_keeps_connection(&refusal));
        let failure = Err(ClientError::Transport {
            kind: TransportErrorKind::Read,
            disposition: SubmissionDisposition::Indeterminate,
        });
        assert!(!response_keeps_connection(&failure));
    }

    #[test]
    fn connection_idle_age_is_fenced_below_the_gateway_prelude_timeout() {
        let now = tokio::time::Instant::now();
        assert!(connection_idle_is_reusable(now - MAX_REUSABLE_CONNECTION_IDLE, now));
        assert!(!connection_idle_is_reusable(
            now - MAX_REUSABLE_CONNECTION_IDLE - Duration::from_millis(1),
            now
        ));
    }

    #[test]
    fn verified_session_cache_is_bounded_and_never_crosses_principals() {
        let metrics = CellTransportMetrics::default();
        let mut cache = VerifiedSessionCache::default();
        let first = cache_key(1, 10, 100);
        let other_principal = cache_key(1, 11, 100);
        assert_ne!(first, other_principal);
        cache.insert(first.clone(), verification(1, 10), &metrics);
        assert!(cache.get(&other_principal, &metrics).is_none());
        assert_eq!(cache.get(&first, &metrics), Some(verification(1, 10)));

        for offset in 0..=MAX_VERIFIED_SESSION_METADATA {
            let identity = u128::try_from(offset).unwrap_or_default().saturating_add(1_000);
            cache.insert(
                cache_key(identity, identity, identity),
                verification(identity, identity),
                &metrics,
            );
        }
        assert_eq!(cache.entries.len(), MAX_VERIFIED_SESSION_METADATA);
        assert!(metrics.session_cache_evictions.load(Ordering::Relaxed) > 0);
        assert_eq!(
            metrics.session_cache_entries.load(Ordering::Relaxed),
            MAX_VERIFIED_SESSION_METADATA
        );
        assert_eq!(metrics.session_cache_hits.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.session_cache_misses.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn generated_multi_index_scope_requires_a_nonempty_distinct_set() {
        fn index(name: &str, incarnation: &str) -> Resource {
            Resource {
                kind: "index".to_owned(),
                name: name.to_owned(),
                incarnation: Identity {
                    value: incarnation.to_owned(),
                },
            }
        }

        fn request(indexes: Vec<Resource>) -> RequestBody {
            RequestBody::OpenPit(OpenPitRequest {
                indexes,
                keep_alive_millis: U64 {
                    value: "1000".to_owned(),
                },
            })
        }

        let first = index("orders", "00000000000000000000000000000001");
        let second = index("events", "00000000000000000000000000000002");
        assert!(matches!(
            resolve_index_name(&request(Vec::new()), None),
            Err(ClientError::CollectionIndexRefRequired)
        ));
        assert!(matches!(
            resolve_index_name(&request(vec![first.clone(), second.clone()]), None),
            Ok(None)
        ));
        assert!(matches!(
            resolve_index_name(&request(vec![first.clone(), first.clone()]), None),
            Err(ClientError::CollectionIndexRefMismatch)
        ));
        assert!(matches!(
            resolve_index_name(&request(vec![second.clone(), first.clone(), first]), None),
            Err(ClientError::CollectionIndexRefMismatch)
        ));
    }

    #[test]
    fn generated_get_index_is_collection_scoped() {
        let request = RequestBody::GetIndex(GetIndexRequest {
            name: "events".to_owned(),
            expected_incarnation: None,
        });

        assert!(matches!(resolve_index_name(&request, None), Ok(None)));
        assert!(matches!(
            resolve_index_name(&request, Some("events")),
            Err(ClientError::CollectionIndexRefMismatch)
        ));
    }
}