asupersync 0.4.10

//! gRPC client implementation.
//!
//! Provides client-side infrastructure for calling gRPC services.

use std::any::Any;
use std::collections::VecDeque;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::task::{Context, Poll, Waker};
use std::time::Duration;

use crate::bytes::{Bytes, BytesMut};

#[cfg(not(target_arch = "wasm32"))]
use std::io;
#[cfg(not(target_arch = "wasm32"))]
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};

#[cfg(not(target_arch = "wasm32"))]
use base64::Engine as _;

#[cfg(not(target_arch = "wasm32"))]
use crate::codec::Decoder as _;
#[cfg(not(target_arch = "wasm32"))]
use crate::http::h2::connection::{CLIENT_PREFACE, ReceivedFrame};
#[cfg(not(target_arch = "wasm32"))]
use crate::http::h2::{Connection, FrameCodec, Header, SettingsBuilder};
#[cfg(not(target_arch = "wasm32"))]
use crate::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _, ReadBuf};
#[cfg(not(target_arch = "wasm32"))]
use crate::net::TcpStream;
use crate::tls::TlsConnector;
#[cfg(all(not(target_arch = "wasm32"), feature = "tls"))]
use crate::tls::TlsStream;

use super::codec::{Codec, FramedCodec, IdentityCodec};
use super::status::{Code, GrpcError, Status, TransportErrorKind};
use super::streaming::{
    MAX_STREAM_BUFFERED, Metadata, MetadataValue, Request, Response, Streaming,
};

/// Supported gRPC message compression encodings for channel negotiation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompressionEncoding {
    /// No compression.
    Identity,
    /// Gzip compression.
    Gzip,
}

impl CompressionEncoding {
    fn as_header_value(self) -> &'static str {
        match self {
            Self::Identity => "identity",
            Self::Gzip => "gzip",
        }
    }

    /// Parse a compression encoding from the `grpc-encoding` header value.
    #[must_use]
    pub fn from_header_value(value: &str) -> Option<Self> {
        match value {
            "identity" => Some(Self::Identity),
            "gzip" => Some(Self::Gzip),
            _ => None,
        }
    }

    /// Return the frame compressor for this encoding, if any.
    ///
    /// Returns `None` for `Identity` (no compression needed).
    /// Requires the `compression` feature for `Gzip`.
    #[must_use]
    pub fn frame_compressor(self) -> Option<super::codec::FrameCompressor> {
        match self {
            Self::Identity => None,
            #[cfg(feature = "compression")]
            Self::Gzip => Some(super::codec::gzip_frame_compress),
            #[cfg(not(feature = "compression"))]
            Self::Gzip => None,
        }
    }

    /// Return the frame decompressor for this encoding, if any.
    ///
    /// Returns `None` for `Identity` (no decompression needed).
    /// Requires the `compression` feature for `Gzip`.
    #[must_use]
    pub fn frame_decompressor(self) -> Option<super::codec::FrameDecompressor> {
        match self {
            Self::Identity => None,
            #[cfg(feature = "compression")]
            Self::Gzip => Some(super::codec::gzip_frame_decompress),
            #[cfg(not(feature = "compression"))]
            Self::Gzip => None,
        }
    }
}

fn effective_send_compression(config: &ChannelConfig) -> Option<CompressionEncoding> {
    match config.send_compression {
        Some(CompressionEncoding::Identity) => Some(CompressionEncoding::Identity),
        Some(encoding) if encoding.frame_compressor().is_some() => Some(encoding),
        _ => None,
    }
}

fn effective_accept_compressions(config: &ChannelConfig) -> Vec<CompressionEncoding> {
    let mut encodings = Vec::new();
    for encoding in &config.accept_compression {
        let supported = matches!(encoding, CompressionEncoding::Identity)
            || encoding.frame_decompressor().is_some();
        if supported && !encodings.contains(encoding) {
            encodings.push(*encoding);
        }
    }
    encodings
}

fn client_framed_codec<C: Codec>(channel: &Channel, codec: C) -> FramedCodec<C> {
    let send_compression = effective_send_compression(channel.config());
    let compressor = send_compression.and_then(CompressionEncoding::frame_compressor);
    let decompressor = effective_accept_compressions(channel.config())
        .into_iter()
        .find(|encoding| *encoding != CompressionEncoding::Identity)
        .and_then(CompressionEncoding::frame_decompressor);

    FramedCodec::with_message_size_limits(
        codec,
        channel.config().max_send_message_size,
        channel.config().max_recv_message_size,
    )
    .with_frame_hooks(compressor, decompressor)
}

/// gRPC channel configuration.
#[derive(Debug, Clone)]
pub struct ChannelConfig {
    /// Connection timeout.
    pub connect_timeout: Duration,
    /// Request timeout (deadline).
    pub timeout: Option<Duration>,
    /// Maximum message size for receiving.
    pub max_recv_message_size: usize,
    /// Maximum message size for sending.
    pub max_send_message_size: usize,
    /// Initial connection window size.
    pub initial_connection_window_size: u32,
    /// Initial stream window size.
    pub initial_stream_window_size: u32,
    /// Keep-alive interval.
    pub keepalive_interval: Option<Duration>,
    /// Keep-alive timeout.
    pub keepalive_timeout: Option<Duration>,
    /// Whether to use TLS.
    pub use_tls: bool,
    /// Compression used for outbound messages.
    pub send_compression: Option<CompressionEncoding>,
    /// Compression encodings accepted by this client.
    pub accept_compression: Vec<CompressionEncoding>,
}

impl Default for ChannelConfig {
    fn default() -> Self {
        Self {
            connect_timeout: Duration::from_secs(5),
            timeout: None,
            max_recv_message_size: 4 * 1024 * 1024,
            max_send_message_size: 4 * 1024 * 1024,
            initial_connection_window_size: 1024 * 1024,
            initial_stream_window_size: 1024 * 1024,
            keepalive_interval: None,
            keepalive_timeout: None,
            use_tls: false,
            send_compression: None,
            accept_compression: vec![CompressionEncoding::Identity],
        }
    }
}

/// Builder for creating a gRPC channel.
#[derive(Debug)]
pub struct ChannelBuilder {
    /// The target URI.
    uri: String,
    /// Channel configuration.
    config: ChannelConfig,
    /// Explicit TLS authority for HTTPS channels.
    tls_connector: Option<TlsConnector>,
    /// Optional certificate identity when it differs from the logical URI host.
    tls_server_name: Option<String>,
    /// Explicit native TCP destination, separate from the logical authority.
    #[cfg(not(target_arch = "wasm32"))]
    dial_addr: Option<SocketAddr>,
}

impl ChannelBuilder {
    /// Create a new channel builder for the given URI.
    ///
    /// The client accepts deterministic in-memory `loopback` targets and
    /// native HTTP/2 `localhost` / `127.0.0.1` targets. Network I/O is lazy:
    /// [`ChannelBuilder::connect`] validates and stores the target, while an
    /// RPC method establishes its transport.
    #[must_use]
    pub fn new(uri: impl Into<String>) -> Self {
        Self {
            uri: uri.into(),
            config: ChannelConfig::default(),
            tls_connector: None,
            tls_server_name: None,
            #[cfg(not(target_arch = "wasm32"))]
            dial_addr: None,
        }
    }

    /// Set the connection timeout.
    #[must_use]
    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
        self.config.connect_timeout = timeout;
        self
    }

    /// Set the request timeout (deadline).
    #[must_use]
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.config.timeout = Some(timeout);
        self
    }

    /// Set the maximum receive message size.
    #[must_use]
    pub fn max_recv_message_size(mut self, size: usize) -> Self {
        self.config.max_recv_message_size = size;
        self
    }

    /// Set the maximum send message size.
    #[must_use]
    pub fn max_send_message_size(mut self, size: usize) -> Self {
        self.config.max_send_message_size = size;
        self
    }

    /// Set the initial connection window size.
    #[must_use]
    pub fn initial_connection_window_size(mut self, size: u32) -> Self {
        self.config.initial_connection_window_size = size;
        self
    }

    /// Set the initial stream window size.
    #[must_use]
    pub fn initial_stream_window_size(mut self, size: u32) -> Self {
        self.config.initial_stream_window_size = size;
        self
    }

    /// Set the keep-alive interval.
    #[must_use]
    pub fn keepalive_interval(mut self, interval: Duration) -> Self {
        self.config.keepalive_interval = Some(interval);
        self
    }

    /// Set the keep-alive timeout.
    #[must_use]
    pub fn keepalive_timeout(mut self, timeout: Duration) -> Self {
        self.config.keepalive_timeout = Some(timeout);
        self
    }

    /// Set the outbound compression encoding.
    #[must_use]
    pub fn send_compression(mut self, encoding: CompressionEncoding) -> Self {
        self.config.send_compression = Some(encoding);
        self
    }

    /// Add one accepted compression encoding.
    #[must_use]
    pub fn accept_compression(mut self, encoding: CompressionEncoding) -> Self {
        self.config.accept_compression.push(encoding);
        self
    }

    /// Replace accepted compression encodings.
    #[must_use]
    pub fn accept_compressions(
        mut self,
        encodings: impl IntoIterator<Item = CompressionEncoding>,
    ) -> Self {
        self.config.accept_compression.clear();
        self.config.accept_compression.extend(encodings);
        self
    }

    /// Require a TLS-backed gRPC transport.
    ///
    /// This only marks the transport requirement. Callers must also provide
    /// an explicit [`TlsConnector`] with [`Self::tls_connector`]; channels
    /// fail closed rather than selecting ambient trust roots.
    #[must_use]
    pub fn tls(mut self) -> Self {
        self.config.use_tls = true;
        self
    }

    /// Supply the TLS connector used by HTTPS unary calls.
    ///
    /// The connector should advertise HTTP/2 (normally via
    /// [`crate::tls::TlsConnectorBuilder::alpn_grpc`]). The gRPC transport
    /// independently verifies that the peer actually negotiated `h2` before
    /// writing the connection preface.
    #[must_use]
    pub fn tls_connector(mut self, connector: TlsConnector) -> Self {
        self.config.use_tls = true;
        self.tls_connector = Some(connector);
        self
    }

    /// Override the DNS name authenticated by TLS.
    ///
    /// This changes only certificate authentication. It never changes the URI
    /// `:authority` or the native socket selected by the localhost default or
    /// `Self::dial_addr`.
    #[must_use]
    pub fn tls_server_name(mut self, server_name: impl Into<String>) -> Self {
        self.config.use_tls = true;
        self.tls_server_name = Some(server_name.into());
        self
    }

    /// Dial one explicit native TCP address while retaining the URI authority.
    ///
    /// This is the capability-safe alternative to ambient DNS. An explicit
    /// address is accepted only for an HTTPS channel with a caller-supplied
    /// [`TlsConnector`]. The URI still supplies HTTP/2 `:authority` and, unless
    /// [`Self::tls_server_name`] overrides it, the certificate identity.
    /// Deterministic `loopback` channels cannot use this native transport.
    #[cfg(not(target_arch = "wasm32"))]
    #[must_use]
    pub fn dial_addr(mut self, address: SocketAddr) -> Self {
        self.dial_addr = Some(address);
        self
    }

    /// Build the channel.
    pub async fn connect(self) -> Result<Channel, GrpcError> {
        Channel::connect_with_transport(
            &self.uri,
            self.config,
            self.tls_connector,
            self.tls_server_name,
            #[cfg(not(target_arch = "wasm32"))]
            self.dial_addr,
        )
        .await
    }
}

/// A gRPC channel representing an explicit client transport capability.
///
/// `loopback` selects deterministic in-memory behavior. `localhost` and
/// `127.0.0.1` select native HTTP/2 over TCP, optionally protected by a
/// caller-supplied TLS connector. `ChannelBuilder::dial_addr` can instead
/// select one explicit native socket for an authenticated HTTPS authority.
/// The channel is intentionally lazy: constructing it performs validation but
/// does not open a socket.
#[derive(Debug, Clone)]
pub struct Channel {
    /// The target URI.
    uri: String,
    /// Channel configuration.
    config: ChannelConfig,
    /// Explicit TLS authority for HTTPS unary calls.
    #[cfg(not(target_arch = "wasm32"))]
    tls_connector: Option<TlsConnector>,
    /// Optional certificate identity distinct from the logical URI host.
    #[cfg(not(target_arch = "wasm32"))]
    tls_server_name: Option<String>,
    /// Explicit native TCP destination distinct from the logical authority.
    #[cfg(not(target_arch = "wasm32"))]
    dial_addr: Option<SocketAddr>,
}

impl Channel {
    /// Create a channel builder for the given URI.
    #[must_use]
    pub fn builder(uri: impl Into<String>) -> ChannelBuilder {
        ChannelBuilder::new(uri)
    }

    /// Connect to a gRPC client transport at the given URI.
    ///
    /// Supports both in-memory loopback transport (host: `loopback`) and real
    /// HTTP/2 connections to localhost (host: `localhost` or `127.0.0.1`).
    /// The first network-backed RPC performs the TCP connection.
    pub async fn connect(uri: impl Into<String>) -> Result<Self, GrpcError> {
        Self::connect_with_config(&uri.into(), ChannelConfig::default()).await
    }

    /// Connect with custom configuration.
    ///
    /// Supports both in-memory loopback transport (host: `loopback`) and real
    /// HTTP/2 connections to localhost (host: `localhost` or `127.0.0.1`).
    /// The first network-backed RPC performs the TCP connection.
    #[allow(clippy::unused_async)]
    pub async fn connect_with_config(uri: &str, config: ChannelConfig) -> Result<Self, GrpcError> {
        Self::connect_with_transport(
            uri,
            config,
            None,
            None,
            #[cfg(not(target_arch = "wasm32"))]
            None,
        )
        .await
    }

    #[allow(clippy::unused_async)]
    async fn connect_with_transport(
        uri: &str,
        config: ChannelConfig,
        tls_connector: Option<TlsConnector>,
        tls_server_name: Option<String>,
        #[cfg(not(target_arch = "wasm32"))] dial_addr: Option<SocketAddr>,
    ) -> Result<Self, GrpcError> {
        #[cfg(not(target_arch = "wasm32"))]
        let has_explicit_dial_addr = dial_addr.is_some();
        #[cfg(target_arch = "wasm32")]
        let has_explicit_dial_addr = false;
        validate_channel_uri(uri, has_explicit_dial_addr)?;
        validate_channel_security(
            uri,
            &config,
            tls_connector.is_some(),
            has_explicit_dial_addr,
        )?;
        if has_explicit_dial_addr {
            let uri_server_name = channel_uri_host(uri).ok_or_else(|| {
                GrpcError::transport_kind(
                    TransportErrorKind::ProtocolViolation,
                    "explicit gRPC dial URI is missing a TLS server identity",
                )
            })?;
            TlsConnector::validate_domain(uri_server_name).map_err(|error| {
                GrpcError::transport_kind(
                    TransportErrorKind::ProtocolViolation,
                    format!("invalid gRPC URI host for explicit TLS dial: {error}"),
                )
            })?;
        }
        if let Some(server_name) = tls_server_name.as_deref() {
            TlsConnector::validate_domain(server_name).map_err(|error| {
                GrpcError::transport_kind(
                    TransportErrorKind::ProtocolViolation,
                    format!("invalid gRPC TLS server name: {error}"),
                )
            })?;
        }
        #[cfg(target_arch = "wasm32")]
        let _ = (&tls_connector, &tls_server_name);
        Ok(Self {
            uri: uri.to_string(),
            config,
            #[cfg(not(target_arch = "wasm32"))]
            tls_connector,
            #[cfg(not(target_arch = "wasm32"))]
            tls_server_name,
            #[cfg(not(target_arch = "wasm32"))]
            dial_addr,
        })
    }

    /// Get the target URI.
    #[must_use]
    pub fn uri(&self) -> &str {
        &self.uri
    }

    /// Get the channel configuration.
    #[must_use]
    pub fn config(&self) -> &ChannelConfig {
        &self.config
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn tls_connector(&self) -> Option<&TlsConnector> {
        self.tls_connector.as_ref()
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn tls_server_name(&self) -> Option<&str> {
        self.tls_server_name.as_deref()
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn dial_addr(&self) -> Option<SocketAddr> {
        self.dial_addr
    }
}

/// A gRPC client for making RPC calls.
pub struct GrpcClient<C = IdentityCodec> {
    /// The underlying channel.
    channel: Channel,
    /// The codec for message serialization.
    codec: FramedCodec<C>,
    /// Client interceptor chain.
    client_interceptors: Vec<Arc<dyn ClientInterceptor>>,
}

impl<C: fmt::Debug> fmt::Debug for GrpcClient<C> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GrpcClient")
            .field("channel", &self.channel)
            .field("codec", &self.codec)
            .field(
                "client_interceptors",
                &format!("[{} interceptors]", self.client_interceptors.len()),
            )
            .finish()
    }
}

impl GrpcClient<IdentityCodec> {
    /// Create a new client with an identity codec.
    #[must_use]
    pub fn new(channel: Channel) -> Self {
        let framed_codec = client_framed_codec(&channel, IdentityCodec);
        Self {
            channel,
            codec: framed_codec,
            client_interceptors: Vec::new(),
        }
    }
}

impl<C: Codec> GrpcClient<C> {
    /// Create a new client with a custom codec.
    #[must_use]
    pub fn with_codec(channel: Channel, codec: C) -> Self {
        let framed_codec = client_framed_codec(&channel, codec);
        Self {
            channel,
            codec: framed_codec,
            client_interceptors: Vec::new(),
        }
    }

    /// Get the underlying channel.
    pub fn channel(&self) -> &Channel {
        &self.channel
    }

    /// Add one client interceptor and return the updated client.
    #[must_use]
    pub fn with_interceptor<I>(mut self, interceptor: I) -> Self
    where
        I: ClientInterceptor + 'static,
    {
        self.client_interceptors.push(Arc::new(interceptor));
        self
    }

    /// Add multiple client interceptors and return the updated client.
    #[must_use]
    pub fn with_interceptors<I>(mut self, interceptors: impl IntoIterator<Item = I>) -> Self
    where
        I: ClientInterceptor + 'static,
    {
        let interceptors = interceptors.into_iter();
        let (lower, upper) = interceptors.size_hint();
        self.client_interceptors.reserve(upper.unwrap_or(lower));
        for interceptor in interceptors {
            self.client_interceptors.push(Arc::new(interceptor));
        }
        self
    }

    /// Register one client interceptor in place.
    pub fn add_interceptor<I>(&mut self, interceptor: I)
    where
        I: ClientInterceptor + 'static,
    {
        self.client_interceptors.push(Arc::new(interceptor));
    }

    /// Returns the number of registered client interceptors.
    #[must_use]
    pub fn interceptor_count(&self) -> usize {
        self.client_interceptors.len()
    }

    fn build_outbound_metadata<Req>(
        &self,
        request: &Request<Req>,
        path: &str,
    ) -> Result<Metadata, Status> {
        let mut metadata_request = Request::with_metadata(Bytes::new(), request.metadata().clone());
        self.apply_channel_metadata_defaults(metadata_request.metadata_mut());
        self.apply_client_interceptors(&mut metadata_request)?;

        let mut metadata = metadata_request.metadata().clone();
        let _ = metadata.insert("x-asupersync-grpc-path", path);
        let transport = if channel_target_is_loopback(self.channel.uri()) {
            "loopback"
        } else {
            "native-h2"
        };
        let _ = metadata.insert("x-asupersync-grpc-transport", transport);
        Ok(metadata)
    }

    fn apply_channel_metadata_defaults(&self, metadata: &mut Metadata) {
        /// br-asupersync-20occs: classification of an existing
        /// `grpc-timeout` entry for the scrub-vs-replace decision.
        enum ExistingTimeoutState {
            Parseable(String),
            Malformed,
            Absent,
        }

        // br-asupersync-20occs: classify the existing entry into one of three
        // states — present-and-parseable, present-but-malformed, or absent.
        // If malformed, scrub it before deciding whether to insert the
        // channel default; previously the malformed value rode through to the
        // wire when channel.config.timeout was None.
        let existing_state = match metadata.get("grpc-timeout") {
            Some(super::streaming::MetadataValue::Ascii(existing))
                if super::server::parse_grpc_timeout(existing).is_some() =>
            {
                ExistingTimeoutState::Parseable(existing.clone())
            }
            Some(_) => ExistingTimeoutState::Malformed,
            None => ExistingTimeoutState::Absent,
        };
        let (timeout_value, base_source) = match existing_state {
            ExistingTimeoutState::Parseable(v) => (Some(v), "override"),
            ExistingTimeoutState::Malformed => {
                // Scrub the malformed entry; fall back to the channel default
                // (which may also be None, in which case no grpc-timeout is
                // sent — equivalent to "no deadline").
                metadata.remove("grpc-timeout");
                (
                    self.channel.config.timeout.map(encode_grpc_timeout),
                    "channel",
                )
            }
            ExistingTimeoutState::Absent => (
                self.channel.config.timeout.map(encode_grpc_timeout),
                "channel",
            ),
        };

        // br-asupersync-server-stack-hardening-eeexl1.1.3: meet the base
        // timeout (per-request override or channel default) with the
        // remaining ambient Cx budget, so an outbound call can never
        // outlive its caller's deadline. Meet semantics: the budget can
        // only tighten the base; with no base, the remaining budget alone
        // becomes the wire deadline; with neither, no grpc-timeout is
        // sent. The original header string is preserved verbatim unless
        // the budget actually tightens it.
        let base_source = if timeout_value.is_some() {
            base_source
        } else {
            "none"
        };
        let ambient_remaining = ambient_remaining_budget();
        let timeout_value = match (timeout_value, ambient_remaining) {
            (Some(base), Some(remaining)) => {
                let base_duration = super::server::parse_grpc_timeout(&base)
                    .expect("base grpc-timeout was validated or encoded above");
                if remaining < base_duration {
                    Some(encode_grpc_timeout(remaining))
                } else {
                    Some(base)
                }
            }
            (None, Some(remaining)) => Some(encode_grpc_timeout(remaining)),
            (base, None) => base,
        };
        if let Some(timeout_value) = timeout_value {
            // Forwarded-budget trace event at the client hop: which base
            // applied (per-request override / channel default / none),
            // the remaining ambient budget, and the wire value sent.
            if let Some(cx) = crate::cx::Cx::current() {
                let remaining = ambient_remaining
                    .map_or_else(|| "none".to_string(), |r| r.as_nanos().to_string());
                cx.trace(&format!(
                    "client.budget_forwarded proto=grpc base={base_source} remaining_ns={remaining} grpc_timeout={timeout_value}",
                ));
            }
            let _ = metadata.insert_or_replace("grpc-timeout", timeout_value);
        }

        if metadata.get("grpc-encoding").is_none()
            && let Some(encoding) = effective_send_compression(self.channel.config())
        {
            let _ = metadata.insert("grpc-encoding", encoding.as_header_value());
        }

        let accept_compression = effective_accept_compressions(self.channel.config());
        if metadata.get("grpc-accept-encoding").is_none() && !accept_compression.is_empty() {
            let encodings = accept_compression
                .iter()
                .map(|encoding| encoding.as_header_value())
                .collect::<Vec<_>>()
                .join(",");
            let _ = metadata.insert("grpc-accept-encoding", encodings);
        }
    }

    fn apply_client_interceptors(&self, request: &mut Request<Bytes>) -> Result<(), Status> {
        for interceptor in &self.client_interceptors {
            interceptor.intercept(request)?;
        }
        Ok(())
    }

    /// Make a unary RPC call.
    pub async fn unary<Req, Resp>(
        &mut self,
        path: &str,
        request: Request<Req>,
    ) -> Result<Response<Resp>, Status>
    where
        Req: Send + 'static,
        Resp: Send + 'static,
    {
        validate_rpc_path(path)?;
        enforce_deadline_budget(self.channel.config.timeout)?;

        let metadata = self.build_outbound_metadata(&request, path)?;
        if channel_target_is_loopback(self.channel.uri()) {
            let payload = convert_message::<Req, Resp>(request.into_inner(), "unary call")?;
            return Ok(Response::with_metadata(payload, metadata));
        }

        #[cfg(target_arch = "wasm32")]
        {
            let _ = request;
            return Err(Status::unimplemented(
                "native HTTP/2 gRPC client transport is unavailable on wasm32",
            ));
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            let typed_request = downcast_boxed_message::<C::Encode>(
                Box::new(request.into_inner()),
                "native HTTP/2 unary request encoding",
            )?;
            let mut framed_request = BytesMut::new();
            self.codec
                .encode_message(&typed_request, &mut framed_request)
                .map_err(GrpcError::into_status)?;

            let wire_response =
                native_h2_unary(&self.channel, path, metadata, framed_request.freeze()).await?;

            let mut framed_response = BytesMut::from(wire_response.body.as_ref());
            let decoded = self
                .codec
                .decode_message_with_encoding(
                    &mut framed_response,
                    wire_response.grpc_encoding.as_deref(),
                )
                .map_err(GrpcError::into_status)?
                .ok_or_else(|| Status::internal("gRPC unary response contained no message"))?;
            if !framed_response.is_empty() {
                return Err(Status::internal(
                    "gRPC unary response contained multiple or trailing message bytes",
                ));
            }
            let payload = downcast_boxed_message::<Resp>(
                Box::new(decoded),
                "native HTTP/2 unary response decoding",
            )?;
            Ok(Response::with_metadata(payload, wire_response.metadata))
        }
    }

    /// Start a server streaming RPC call.
    #[allow(clippy::unused_async)]
    pub async fn server_streaming<Req, Resp>(
        &mut self,
        path: &str,
        request: Request<Req>,
    ) -> Result<Response<ResponseStream<Resp>>, Status>
    where
        Req: Send + 'static,
        Resp: Send + 'static,
    {
        validate_rpc_path(path)?;
        enforce_deadline_budget(self.channel.config.timeout)?;

        if !channel_target_is_loopback(self.channel.uri()) {
            return Err(Status::unimplemented(
                "native HTTP/2 server-streaming client transport is not wired yet",
            ));
        }

        let metadata = self.build_outbound_metadata(&request, path)?;
        let mut stream = ResponseStream::open();
        let payload = convert_message::<Req, Resp>(request.into_inner(), "server streaming call")?;
        stream.push(Ok(payload))?;
        stream.close();

        Ok(Response::with_metadata(stream, metadata))
    }

    /// Start a client streaming RPC call.
    #[allow(clippy::unused_async)]
    pub async fn client_streaming<Req, Resp>(
        &mut self,
        path: &str,
    ) -> Result<(RequestSink<Req>, ResponseFuture<Resp>), Status>
    where
        Req: Send + 'static,
        Resp: Send + 'static,
    {
        validate_rpc_path(path)?;
        enforce_deadline_budget(self.channel.config.timeout)?;

        if !channel_target_is_loopback(self.channel.uri()) {
            return Err(Status::unimplemented(
                "native HTTP/2 client-streaming client transport is not wired yet",
            ));
        }

        let request = Request::new(Bytes::new());
        let metadata = self.build_outbound_metadata(&request, path)?;
        let state = Arc::new(Mutex::new(RequestSinkState::new()));
        let sink = RequestSink::from_state(state.clone());
        let future = ResponseFuture::with_resolver(state, move |state| {
            if state.sent_count > 1 {
                return Err(Status::failed_precondition(
                    "loopback client streaming does not support multiple request messages yet",
                ));
            }
            let Some(last) = state.last_message.take() else {
                return Err(Status::invalid_argument(
                    "client stream closed without any request messages",
                ));
            };
            let response =
                downcast_boxed_message::<Resp>(last, "client streaming response conversion")?;
            Ok(Response::with_metadata(response, metadata.clone()))
        });
        Ok((sink, future))
    }

    /// Start a bidirectional streaming RPC call.
    #[allow(clippy::unused_async)]
    pub async fn bidi_streaming<Req, Resp>(
        &mut self,
        path: &str,
    ) -> Result<(RequestSink<Req>, ResponseStream<Resp>), Status>
    where
        Req: Send + 'static,
        Resp: Send + 'static,
    {
        validate_rpc_path(path)?;
        enforce_deadline_budget(self.channel.config.timeout)?;

        if !channel_target_is_loopback(self.channel.uri()) {
            return Err(Status::unimplemented(
                "native HTTP/2 bidirectional-streaming client transport is not wired yet",
            ));
        }

        let request = Request::new(Bytes::new());
        let _metadata = self.build_outbound_metadata(&request, path)?;
        let request_state = Arc::new(Mutex::new(RequestSinkState::new()));
        let cancel_request_state = Arc::clone(&request_state);
        let stream = ResponseStream::open_with_cancel_hook(Box::new(move || {
            cancel_request_sink_state(
                &cancel_request_state,
                Status::cancelled("response stream cancelled by client"),
            );
            Ok(())
        }));
        let mut send_stream = stream.clone();
        let close_stream = stream.clone();
        let cancel_stream = stream.clone();
        let sink = RequestSink::with_state_and_hooks(
            request_state,
            Some(Box::new(move |message: Req| {
                let response =
                    convert_message::<Req, Resp>(message, "bidirectional streaming conversion")?;
                send_stream.push(Ok(response))
            })),
            Some(Box::new(move || {
                close_stream.close();
                Ok(())
            })),
            Some(Box::new(move || {
                cancel_stream.cancel(Status::cancelled("request stream cancelled by client"));
                Ok(())
            })),
        );
        Ok((sink, stream))
    }
}

fn channel_target_is_loopback(uri: &str) -> bool {
    let Some((_, remainder)) = uri.split_once("://") else {
        return false;
    };
    let authority = remainder.split(['/', '?', '#']).next().unwrap_or_default();
    let host_port = authority
        .rsplit_once('@')
        .map_or(authority, |(_, value)| value);
    let host = host_port
        .split_once(':')
        .map_or(host_port, |(value, _)| value);
    host.eq_ignore_ascii_case("loopback")
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug)]
struct NativeH2Target {
    authority: String,
    address: SocketAddr,
    server_name: String,
    scheme: &'static str,
    use_tls: bool,
}

#[cfg(not(target_arch = "wasm32"))]
impl NativeH2Target {
    fn parse(
        uri: &str,
        use_tls: bool,
        tls_server_name: Option<&str>,
        dial_addr: Option<SocketAddr>,
    ) -> Result<Self, Status> {
        let (uri_scheme, remainder) = uri
            .split_once("://")
            .ok_or_else(|| Status::unavailable("channel URI is missing a scheme separator"))?;
        let use_tls = use_tls || uri_scheme.eq_ignore_ascii_case("https");
        let authority = remainder
            .split(['/', '?', '#'])
            .next()
            .ok_or_else(|| Status::unavailable("channel URI is missing an authority"))?;
        if authority.contains('@') {
            return Err(Status::unavailable(
                "userinfo is not supported by the native HTTP/2 gRPC client",
            ));
        }
        let (host, port) = match authority.rsplit_once(':') {
            Some((host, port)) => {
                let port = port.parse::<u16>().map_err(|_| {
                    Status::unavailable("channel URI port must be an unsigned 16-bit integer")
                })?;
                (host, port)
            }
            None => (authority, if use_tls { 443 } else { 80 }),
        };
        let address = if let Some(address) = dial_addr {
            if host.eq_ignore_ascii_case("loopback") {
                return Err(Status::failed_precondition(
                    "deterministic loopback channels cannot use an explicit native dial address",
                ));
            }
            if !use_tls {
                return Err(Status::failed_precondition(
                    "an explicit gRPC dial address requires authenticated HTTPS",
                ));
            }
            address
        } else if host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1" {
            SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port))
        } else {
            return Err(Status::unavailable(
                "native HTTP/2 gRPC transport is restricted to localhost",
            ));
        };
        Ok(Self {
            authority: authority.to_ascii_lowercase(),
            address,
            server_name: tls_server_name.map_or_else(
                || host.to_ascii_lowercase(),
                |name| name.to_ascii_lowercase(),
            ),
            scheme: if use_tls { "https" } else { "http" },
            use_tls,
        })
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug)]
enum NativeH2Io {
    Plain(TcpStream),
    #[cfg(feature = "tls")]
    Tls(TlsStream<TcpStream>),
}

#[cfg(not(target_arch = "wasm32"))]
impl AsyncRead for NativeH2Io {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        match &mut *self {
            Self::Plain(stream) => Pin::new(stream).poll_read(cx, buf),
            #[cfg(feature = "tls")]
            Self::Tls(stream) => Pin::new(stream).poll_read(cx, buf),
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl AsyncWrite for NativeH2Io {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        data: &[u8],
    ) -> Poll<io::Result<usize>> {
        match &mut *self {
            Self::Plain(stream) => Pin::new(stream).poll_write(cx, data),
            #[cfg(feature = "tls")]
            Self::Tls(stream) => Pin::new(stream).poll_write(cx, data),
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut *self {
            Self::Plain(stream) => Pin::new(stream).poll_flush(cx),
            #[cfg(feature = "tls")]
            Self::Tls(stream) => Pin::new(stream).poll_flush(cx),
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut *self {
            Self::Plain(stream) => Pin::new(stream).poll_shutdown(cx),
            #[cfg(feature = "tls")]
            Self::Tls(stream) => Pin::new(stream).poll_shutdown(cx),
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug)]
struct NativeUnaryWireResponse {
    body: Bytes,
    metadata: Metadata,
    grpc_encoding: Option<String>,
}

#[cfg(not(target_arch = "wasm32"))]
async fn native_h2_unary(
    channel: &Channel,
    path: &str,
    metadata: Metadata,
    body: Bytes,
) -> Result<NativeUnaryWireResponse, Status> {
    let cx = crate::cx::Cx::current().ok_or_else(|| {
        Status::failed_precondition(
            "native HTTP/2 gRPC calls require an ambient runtime Cx capability",
        )
    })?;
    let target = NativeH2Target::parse(
        channel.uri(),
        channel.config().use_tls,
        channel.tls_server_name(),
        channel.dial_addr(),
    )?;
    let config = channel.config().clone();
    let tls_connector = channel.tls_connector().cloned();
    let timeout = effective_native_call_timeout(&config);
    let now = cx
        .timer_driver()
        .map_or_else(crate::time::wall_now, |timer| timer.now());
    match crate::time::timeout(
        now,
        timeout,
        native_h2_unary_io(target, config, tls_connector, path, &metadata, body),
    )
    .await
    {
        Ok(result) => result,
        Err(_) => Err(Status::deadline_exceeded(format!(
            "native HTTP/2 gRPC call exceeded its {timeout:?} deadline"
        ))),
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn effective_native_call_timeout(config: &ChannelConfig) -> Duration {
    let configured = config.timeout.unwrap_or(config.connect_timeout);
    ambient_remaining_budget().map_or(configured, |remaining| configured.min(remaining))
}

#[cfg(not(target_arch = "wasm32"))]
async fn native_h2_unary_io(
    target: NativeH2Target,
    config: ChannelConfig,
    tls_connector: Option<TlsConnector>,
    path: &str,
    metadata: &Metadata,
    request_body: Bytes,
) -> Result<NativeUnaryWireResponse, Status> {
    let connect_timeout = config.connect_timeout;
    let now = crate::cx::Cx::with_current(|cx| {
        cx.timer_driver()
            .map_or_else(crate::time::wall_now, |timer| timer.now())
    })
    .unwrap_or_else(crate::time::wall_now);
    let mut stream = match crate::time::timeout(now, connect_timeout, async {
        let stream = TcpStream::connect_timeout(target.address, connect_timeout)
            .await
            .map_err(|error| transport_status("connect", error))?;
        native_h2_transport(stream, &target, tls_connector).await
    })
    .await
    {
        Ok(result) => result?,
        Err(_) => {
            return Err(Status::unavailable(format!(
                "native gRPC connection establishment exceeded its {connect_timeout:?} timeout"
            )));
        }
    };

    let settings = SettingsBuilder::client()
        .initial_window_size(config.initial_stream_window_size)
        .build();
    let mut connection = Connection::client(settings);
    // RFC 9113 requires the connection preface's first frame to be SETTINGS.
    // Queue it before expanding the connection receive window, because the
    // latter emits WINDOW_UPDATE when the configured window exceeds 65,535.
    connection.queue_initial_settings();
    connection
        .set_initial_connection_recv_window(config.initial_connection_window_size)
        .map_err(|error| Status::internal(format!("invalid HTTP/2 receive window: {error}")))?;
    let headers = native_h2_request_headers(&target.authority, target.scheme, path, metadata)?;
    let stream_id = connection
        .open_stream(headers, false)
        .map_err(|error| Status::internal(format!("open HTTP/2 request stream: {error}")))?;
    connection
        .send_data(stream_id, request_body, true)
        .map_err(|error| Status::internal(format!("queue HTTP/2 request body: {error}")))?;

    stream
        .write_all(CLIENT_PREFACE)
        .await
        .map_err(|error| transport_status("write HTTP/2 client preface", error))?;
    flush_native_h2_frames(&mut connection, &mut stream).await?;

    let mut codec = FrameCodec::new();
    let mut inbound = BytesMut::new();
    let mut accumulator = NativeUnaryAccumulator::new(config.max_recv_message_size);
    let mut chunk = [0_u8; 16 * 1024];
    loop {
        while let Some(frame) = codec
            .decode(&mut inbound)
            .map_err(|error| Status::internal(format!("decode HTTP/2 frame: {error}")))?
        {
            let received = connection
                .process_frame(frame)
                .map_err(|error| Status::internal(format!("process HTTP/2 frame: {error}")))?;
            if let Some(received) = received {
                accumulator.observe(stream_id, received)?;
            }
            flush_native_h2_frames(&mut connection, &mut stream).await?;
            if accumulator.is_complete() {
                return accumulator.finish();
            }
        }

        let read = stream
            .read(&mut chunk)
            .await
            .map_err(|error| transport_status("read HTTP/2 response", error))?;
        if read == 0 {
            return Err(Status::unavailable(
                "HTTP/2 peer closed before the unary gRPC response completed",
            ));
        }
        inbound.extend_from_slice(&chunk[..read]);
    }
}

#[cfg(not(target_arch = "wasm32"))]
async fn native_h2_transport(
    stream: TcpStream,
    target: &NativeH2Target,
    tls_connector: Option<TlsConnector>,
) -> Result<NativeH2Io, Status> {
    if !target.use_tls {
        return Ok(NativeH2Io::Plain(stream));
    }
    let connector = tls_connector.ok_or_else(|| {
        Status::failed_precondition(
            "HTTPS gRPC transport requires an explicit caller-supplied TLS connector",
        )
    })?;

    #[cfg(feature = "tls")]
    {
        let tls = connector
            .connect(&target.server_name, stream)
            .await
            .map_err(|error| {
                ambient_cancellation_status("gRPC TLS handshake").unwrap_or_else(|| {
                    Status::unavailable(format!("gRPC TLS handshake failed: {error}"))
                })
            })?;
        if tls.alpn_protocol() != Some(b"h2".as_slice()) {
            return Err(Status::unavailable(
                "gRPC TLS peer did not negotiate the required h2 ALPN protocol",
            ));
        }
        Ok(NativeH2Io::Tls(tls))
    }

    #[cfg(not(feature = "tls"))]
    {
        let _ = (stream, connector, target.server_name.as_str());
        Err(Status::unavailable(
            "gRPC TLS support is disabled; rebuild with --features tls",
        ))
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn native_h2_request_headers(
    authority: &str,
    scheme: &str,
    path: &str,
    metadata: &Metadata,
) -> Result<Vec<Header>, Status> {
    let mut headers = vec![
        Header::new(":method", "POST"),
        Header::new(":scheme", scheme),
        Header::new(":path", path),
        Header::new(":authority", authority),
        Header::new("content-type", "application/grpc"),
        Header::new("te", "trailers"),
    ];
    for (name, value) in metadata.iter() {
        if name.starts_with(':')
            || [
                "content-type",
                "te",
                "host",
                "grpc-status",
                "grpc-message",
                "grpc-status-details-bin",
            ]
            .iter()
            .any(|reserved| name.eq_ignore_ascii_case(reserved))
        {
            return Err(Status::invalid_argument(format!(
                "outbound metadata uses transport-reserved key '{name}'"
            )));
        }
        let value = match value {
            MetadataValue::Ascii(value) => value.clone(),
            MetadataValue::Binary(value) => {
                base64::engine::general_purpose::STANDARD_NO_PAD.encode(value)
            }
        };
        headers.push(Header::new(name, value));
    }
    Ok(headers)
}

#[cfg(not(target_arch = "wasm32"))]
async fn flush_native_h2_frames<IO>(
    connection: &mut Connection,
    stream: &mut IO,
) -> Result<(), Status>
where
    IO: AsyncWrite + Unpin,
{
    let mut outbound = BytesMut::new();
    while let Some(frame) = connection.next_frame() {
        frame
            .encode(&mut outbound)
            .map_err(|error| Status::internal(format!("encode HTTP/2 frame: {error}")))?;
    }
    if !outbound.is_empty() {
        stream
            .write_all(&outbound)
            .await
            .map_err(|error| transport_status("write HTTP/2 frames", error))?;
        stream
            .flush()
            .await
            .map_err(|error| transport_status("flush HTTP/2 frames", error))?;
    }
    Ok(())
}

#[cfg(not(target_arch = "wasm32"))]
fn transport_status(context: &str, error: std::io::Error) -> Status {
    if error.kind() == std::io::ErrorKind::Interrupted
        && let Some(status) = ambient_cancellation_status(context)
    {
        return status;
    }
    let kind = TransportErrorKind::from_io_error_kind(error.kind());
    GrpcError::transport_kind(kind, format!("{context}: {error}")).into_status()
}

#[cfg(not(target_arch = "wasm32"))]
fn ambient_cancellation_status(context: &str) -> Option<Status> {
    crate::cx::Cx::with_current(|cx| cx.checkpoint().is_err())
        .unwrap_or(false)
        .then(|| Status::cancelled(format!("{context} cancelled by the caller")))
}

#[cfg(not(target_arch = "wasm32"))]
struct NativeUnaryAccumulator {
    max_body_bytes: usize,
    body: BytesMut,
    metadata: Metadata,
    http_status: Option<u16>,
    content_type_valid: bool,
    grpc_encoding: Option<String>,
    grpc_status: Option<Code>,
    grpc_message: Option<String>,
    grpc_details: Option<Bytes>,
    stream_ended: bool,
}

#[cfg(not(target_arch = "wasm32"))]
impl NativeUnaryAccumulator {
    fn new(max_message_size: usize) -> Self {
        Self {
            max_body_bytes: max_message_size.saturating_add(5),
            body: BytesMut::new(),
            metadata: Metadata::new(),
            http_status: None,
            content_type_valid: false,
            grpc_encoding: None,
            grpc_status: None,
            grpc_message: None,
            grpc_details: None,
            stream_ended: false,
        }
    }

    fn observe(&mut self, stream_id: u32, frame: ReceivedFrame) -> Result<(), Status> {
        match frame {
            ReceivedFrame::Headers {
                stream_id: received_stream,
                headers,
                end_stream,
            } if received_stream == stream_id => {
                self.observe_headers(headers)?;
                self.stream_ended |= end_stream;
            }
            ReceivedFrame::Data {
                stream_id: received_stream,
                data,
                end_stream,
            } if received_stream == stream_id => {
                if self.body.len().saturating_add(data.len()) > self.max_body_bytes {
                    return Err(Status::resource_exhausted(
                        "gRPC unary response exceeds configured receive-message limit",
                    ));
                }
                self.body.extend_from_slice(&data);
                self.stream_ended |= end_stream;
            }
            ReceivedFrame::Reset {
                stream_id: received_stream,
                error_code,
            } if received_stream == stream_id => {
                return Err(Status::from_h2_rst_stream_code(error_code));
            }
            ReceivedFrame::GoAway {
                last_stream_id,
                error_code,
                ..
            } if error_code != crate::http::h2::ErrorCode::NoError
                || last_stream_id < stream_id =>
            {
                return Err(Status::unavailable(format!(
                    "HTTP/2 peer sent GOAWAY before unary response completion: {error_code}"
                )));
            }
            _ => {}
        }
        Ok(())
    }

    fn observe_headers(&mut self, headers: Vec<Header>) -> Result<(), Status> {
        for Header { name, value } in headers {
            if name == ":status" {
                let status = value
                    .parse::<u16>()
                    .map_err(|_| Status::internal("HTTP/2 response has malformed :status"))?;
                if status >= 200 {
                    if self.http_status.replace(status).is_some() {
                        return Err(Status::internal(
                            "HTTP/2 response contains duplicate final :status",
                        ));
                    }
                }
                continue;
            }
            if name.eq_ignore_ascii_case("content-type") {
                self.content_type_valid =
                    value.to_ascii_lowercase().starts_with("application/grpc");
                continue;
            }
            if name.eq_ignore_ascii_case("grpc-encoding") {
                self.grpc_encoding = Some(value);
                continue;
            }
            if name.eq_ignore_ascii_case("grpc-status") {
                if self.grpc_status.is_some() {
                    return Err(Status::internal(
                        "gRPC response contains duplicate grpc-status",
                    ));
                }
                let raw = value
                    .parse::<i32>()
                    .map_err(|_| Status::internal("gRPC response has malformed grpc-status"))?;
                if !(0..=16).contains(&raw) {
                    return Err(Status::internal("gRPC response has unknown grpc-status"));
                }
                self.grpc_status = Some(Code::from_i32(raw));
                continue;
            }
            if name.eq_ignore_ascii_case("grpc-message") {
                if self.grpc_message.is_some() {
                    return Err(Status::internal(
                        "gRPC response contains duplicate grpc-message",
                    ));
                }
                self.grpc_message = Some(
                    super::status::percent_decode_grpc_message(&value)
                        .map_err(GrpcError::into_status)?,
                );
                continue;
            }
            if name.eq_ignore_ascii_case("grpc-status-details-bin") {
                if self.grpc_details.is_some() {
                    return Err(Status::internal(
                        "gRPC response contains duplicate grpc-status-details-bin",
                    ));
                }
                let details = base64::engine::general_purpose::STANDARD
                    .decode(&value)
                    .or_else(|_| base64::engine::general_purpose::STANDARD_NO_PAD.decode(&value))
                    .map_err(|_| Status::internal("gRPC response details are not valid base64"))?;
                self.grpc_details = Some(Bytes::from(details));
                continue;
            }
            if name.starts_with(':') || name.eq_ignore_ascii_case("te") {
                continue;
            }
            let inserted = if name.ends_with("-bin") {
                let decoded = base64::engine::general_purpose::STANDARD
                    .decode(&value)
                    .or_else(|_| base64::engine::general_purpose::STANDARD_NO_PAD.decode(&value))
                    .map_err(|_| {
                        Status::internal(format!(
                            "binary response metadata '{name}' is not valid base64"
                        ))
                    })?;
                self.metadata.insert_bin(name, Bytes::from(decoded))
            } else {
                self.metadata.insert(name, value)
            };
            if !inserted {
                return Err(Status::internal("gRPC response contains invalid metadata"));
            }
        }
        Ok(())
    }

    fn is_complete(&self) -> bool {
        self.stream_ended && self.grpc_status.is_some()
    }

    fn finish(self) -> Result<NativeUnaryWireResponse, Status> {
        if self.http_status != Some(200) {
            return Err(Status::unavailable(format!(
                "gRPC server returned HTTP status {}",
                self.http_status
                    .map_or_else(|| "missing".to_owned(), |status| status.to_string())
            )));
        }
        if !self.content_type_valid {
            return Err(Status::internal(
                "gRPC response is missing a valid application/grpc content-type",
            ));
        }
        let status = self
            .grpc_status
            .ok_or_else(|| Status::internal("gRPC response is missing grpc-status"))?;
        if status != Code::Ok {
            let message = self.grpc_message.unwrap_or_default();
            return Err(match self.grpc_details {
                Some(details) => Status::with_details(status, message, details),
                None => Status::new(status, message),
            });
        }
        Ok(NativeUnaryWireResponse {
            body: self.body.freeze(),
            metadata: self.metadata,
            grpc_encoding: self.grpc_encoding,
        })
    }
}

fn validate_channel_uri(uri: &str, has_explicit_dial_addr: bool) -> Result<(), GrpcError> {
    if uri.is_empty() {
        return Err(GrpcError::transport("channel URI cannot be empty"));
    }
    let (scheme, remainder) = uri
        .split_once("://")
        .ok_or_else(|| GrpcError::transport("channel URI is missing a scheme separator"))?;
    if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
        return Err(GrpcError::transport(
            "channel URI must start with http:// or https://",
        ));
    }
    let authority = remainder
        .split(['/', '?', '#'])
        .next()
        .ok_or_else(|| GrpcError::transport("channel URI is missing an authority"))?;
    if authority
        .chars()
        .any(|ch| ch.is_ascii_whitespace() || ch.is_control())
    {
        return Err(GrpcError::transport(
            "channel URI authority cannot contain whitespace or control characters",
        ));
    }
    if has_explicit_dial_addr && authority.contains('@') {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "userinfo is not supported with an explicit gRPC dial address",
        ));
    }
    // Strip userinfo (RFC 3986 §3.2: authority = [userinfo "@"] host [":" port])
    // before extracting the host, so "loopback:pw@evil.com" doesn't pass.
    let host_port = authority.rsplit_once('@').map_or(authority, |(_, hp)| hp);
    let (host, port) = host_port
        .rsplit_once(':')
        .map_or((host_port, None), |(host, port)| (host, Some(port)));
    if host.is_empty() {
        return Err(GrpcError::transport("channel URI is missing a host"));
    }
    if has_explicit_dial_addr && (host.contains(':') || host.contains('[') || host.contains(']')) {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "explicit gRPC dial authority must contain one DNS or IPv4 host and at most one port",
        ));
    }
    if !has_explicit_dial_addr
        && !host.eq_ignore_ascii_case("loopback")
        && !host.eq_ignore_ascii_case("localhost")
        && host != "127.0.0.1"
    {
        return Err(GrpcError::transport(
            "gRPC client transport supports loopback and localhost only; use a URI with host `loopback`, `localhost`, or `127.0.0.1`",
        ));
    }
    if let Some(port) = port
        && port.parse::<u16>().is_err()
    {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ConnectFailed,
            "channel URI port must be an unsigned 16-bit integer",
        ));
    }
    Ok(())
}

fn channel_uri_host(uri: &str) -> Option<&str> {
    let (_, remainder) = uri.split_once("://")?;
    let authority = remainder.split(['/', '?', '#']).next()?;
    let host_port = authority.rsplit_once('@').map_or(authority, |(_, hp)| hp);
    Some(
        host_port
            .rsplit_once(':')
            .map_or(host_port, |(host, _)| host),
    )
}

fn validate_channel_security(
    uri: &str,
    config: &ChannelConfig,
    has_tls_connector: bool,
    has_explicit_dial_addr: bool,
) -> Result<(), GrpcError> {
    let (scheme, _) = uri
        .split_once("://")
        .ok_or_else(|| GrpcError::transport("channel URI is missing a scheme separator"))?;
    let tls_requested = scheme.eq_ignore_ascii_case("https") || config.use_tls;
    if has_explicit_dial_addr && !scheme.eq_ignore_ascii_case("https") {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "an explicit gRPC dial address requires an https URI",
        ));
    }
    if has_explicit_dial_addr && channel_target_is_loopback(uri) {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "deterministic loopback channels cannot use an explicit native dial address",
        ));
    }
    if tls_requested && channel_target_is_loopback(uri) {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "gRPC TLS is available only on the native localhost transport; \
             deterministic loopback channels do not negotiate TLS",
        ));
    }
    if tls_requested && !has_tls_connector {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "gRPC TLS channel requested without an explicit TLS connector; \
             configure ChannelBuilder::tls_connector so trust roots and ALPN \
             policy remain caller-controlled",
        ));
    }
    if has_tls_connector && !tls_requested {
        return Err(GrpcError::transport_kind(
            TransportErrorKind::ProtocolViolation,
            "gRPC TLS connector supplied for a cleartext channel",
        ));
    }
    Ok(())
}

fn validate_rpc_path(path: &str) -> Result<(), Status> {
    if path.is_empty() {
        return Err(Status::invalid_argument("RPC path cannot be empty"));
    }
    if !path.starts_with('/') {
        return Err(Status::invalid_argument(
            "RPC path must start with '/' (for example: /pkg.Service/Method)",
        ));
    }
    let mut segments = path.split('/');
    let _ = segments.next();
    let service = segments.next();
    let method = segments.next();
    if service.is_none_or(str::is_empty)
        || method.is_none_or(str::is_empty)
        || segments.next().is_some()
    {
        return Err(Status::invalid_argument(
            "RPC path must include service and method segments",
        ));
    }
    Ok(())
}

fn enforce_deadline_budget(timeout: Option<Duration>) -> Result<(), Status> {
    if timeout.is_some_and(|value| value.is_zero()) {
        return Err(Status::deadline_exceeded(
            "configured timeout is zero duration",
        ));
    }
    // br-asupersync-server-stack-hardening-eeexl1.1.3: an outbound call
    // whose caller has already exhausted its budget deadline fails fast
    // locally instead of sending a doomed request.
    if ambient_remaining_budget().is_some_and(|remaining| remaining.is_zero()) {
        return Err(Status::deadline_exceeded(
            "ambient budget deadline already expired",
        ));
    }
    Ok(())
}

/// Remaining time until the ambient [`Cx`](crate::cx::Cx) budget deadline,
/// read against the ambient timer driver (exact under lab virtual time)
/// with a wall-clock fallback (br-asupersync-server-stack-hardening-eeexl1.1.3).
///
/// `None` when no ambient context is installed or its budget carries no
/// deadline; `Some(Duration::ZERO)` when the deadline has already passed.
fn ambient_remaining_budget() -> Option<Duration> {
    crate::cx::Cx::with_current(|cx| {
        let deadline = cx.budget().deadline?;
        let now = cx
            .timer_driver()
            .map_or_else(crate::time::wall_now, |timer| timer.now());
        Some(Duration::from_nanos(deadline.duration_since(now)))
    })
    .flatten()
}

fn encode_grpc_timeout(timeout: Duration) -> String {
    const MAX_GRPC_TIMEOUT_VALUE: u128 = 99_999_999;
    const GRPC_TIMEOUT_UNITS: [(u128, char); 6] = [
        (3_600_000_000_000, 'H'),
        (60_000_000_000, 'M'),
        (1_000_000_000, 'S'),
        (1_000_000, 'm'),
        (1_000, 'u'),
        (1, 'n'),
    ];

    let timeout_nanos = timeout.as_nanos().max(1);

    for &(unit_nanos, suffix) in &GRPC_TIMEOUT_UNITS {
        if timeout_nanos.is_multiple_of(unit_nanos) {
            let value = timeout_nanos / unit_nanos;
            if value <= MAX_GRPC_TIMEOUT_VALUE {
                return format!("{value}{suffix}");
            }
        }
    }

    for &(unit_nanos, suffix) in GRPC_TIMEOUT_UNITS.iter().rev() {
        let value = timeout_nanos.div_ceil(unit_nanos);
        if value <= MAX_GRPC_TIMEOUT_VALUE {
            return format!("{value}{suffix}");
        }
    }
    "99999999H".to_owned()
}

fn convert_message<Req, Resp>(request: Req, context: &str) -> Result<Resp, Status>
where
    Req: Send + 'static,
    Resp: Send + 'static,
{
    downcast_boxed_message::<Resp>(Box::new(request), context)
}

fn downcast_boxed_message<T>(message: Box<dyn Any + Send>, context: &str) -> Result<T, Status>
where
    T: Send + 'static,
{
    message.downcast::<T>().map_or_else(
        |_| {
            Err(Status::failed_precondition(format!(
                "{context} requires matching request/response message types in loopback mode"
            )))
        },
        |value| Ok(*value),
    )
}

fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(PoisonError::into_inner)
}

struct ResponseStreamState<T> {
    items: VecDeque<Result<T, Status>>,
    closed: bool,
    terminal_status: Option<Status>,
    terminal_metadata: Metadata,
    waiters: Vec<Waker>,
    /// Producer wakers parked on [`ResponseStream::poll_reserve`] while the
    /// bounded buffer is full; woken on the next consumer drain or on close.
    producer_waiters: Vec<Waker>,
}

impl<T> fmt::Debug for ResponseStreamState<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ResponseStreamState")
            .field("items_len", &self.items.len())
            .field("closed", &self.closed)
            .field("terminal_status", &self.terminal_status)
            .field("terminal_metadata", &self.terminal_metadata)
            .field("waiters_len", &self.waiters.len())
            .field("producer_waiters_len", &self.producer_waiters.len())
            .finish()
    }
}

impl<T> ResponseStreamState<T> {
    fn closed() -> Self {
        Self {
            items: VecDeque::new(),
            closed: true,
            terminal_status: None,
            terminal_metadata: Metadata::new(),
            waiters: Vec::new(),
            producer_waiters: Vec::new(),
        }
    }

    fn open() -> Self {
        Self {
            items: VecDeque::new(),
            closed: false,
            terminal_status: None,
            terminal_metadata: Metadata::new(),
            waiters: Vec::new(),
            producer_waiters: Vec::new(),
        }
    }

    fn take_waiters(&mut self) -> Vec<Waker> {
        std::mem::take(&mut self.waiters)
    }

    fn register_waiter(&mut self, waker: &Waker) {
        if !self
            .waiters
            .iter()
            .any(|existing| existing.will_wake(waker))
        {
            if self.waiters.len() >= 32 {
                let evicted = self.waiters.remove(0);
                evicted.wake();
            }
            self.waiters.push(waker.clone());
        }
    }

    fn take_producer_waiters(&mut self) -> Vec<Waker> {
        std::mem::take(&mut self.producer_waiters)
    }

    fn register_producer_waiter(&mut self, waker: &Waker) {
        if !self
            .producer_waiters
            .iter()
            .any(|existing| existing.will_wake(waker))
        {
            if self.producer_waiters.len() >= 32 {
                let evicted = self.producer_waiters.remove(0);
                evicted.wake();
            }
            self.producer_waiters.push(waker.clone());
        }
    }
}

/// A stream of responses from the server.
pub struct ResponseStream<T> {
    state: Arc<Mutex<ResponseStreamState<T>>>,
    on_cancel: Option<Arc<Mutex<CloseHook>>>,
}

impl<T> Clone for ResponseStream<T> {
    fn clone(&self) -> Self {
        Self {
            state: Arc::clone(&self.state),
            on_cancel: self.on_cancel.as_ref().map(Arc::clone),
        }
    }
}

impl<T> fmt::Debug for ResponseStream<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ResponseStream")
            .field("state", &self.state)
            .field("has_cancel_hook", &self.on_cancel.is_some())
            .finish()
    }
}

impl<T> ResponseStream<T> {
    /// Create a new response stream.
    #[must_use]
    pub fn new() -> Self {
        Self {
            state: Arc::new(Mutex::new(ResponseStreamState::closed())),
            on_cancel: None,
        }
    }

    /// Create an open response stream that can receive additional items.
    #[must_use]
    pub fn open() -> Self {
        Self {
            state: Arc::new(Mutex::new(ResponseStreamState::open())),
            on_cancel: None,
        }
    }

    fn open_with_cancel_hook(on_cancel: CloseHook) -> Self {
        Self {
            state: Arc::new(Mutex::new(ResponseStreamState::open())),
            on_cancel: Some(Arc::new(Mutex::new(on_cancel))),
        }
    }

    /// Push a response item into the stream.
    ///
    /// Returns an error if the stream has already been closed.
    pub fn push(&mut self, item: Result<T, Status>) -> Result<(), Status> {
        let waiters = {
            let mut state = lock_unpoisoned(&self.state);
            if state.closed {
                return Err(Status::failed_precondition(
                    "cannot push to a closed response stream",
                ));
            }
            if state.items.len() >= MAX_STREAM_BUFFERED {
                return Err(Status::resource_exhausted(
                    "response stream buffer full — apply backpressure",
                ));
            }
            state.items.push_back(item);
            state.take_waiters()
        };
        for waker in waiters {
            waker.wake();
        }
        Ok(())
    }

    /// Number of queued response items currently waiting to be consumed.
    #[must_use]
    pub fn buffer_len(&self) -> usize {
        lock_unpoisoned(&self.state).items.len()
    }

    /// Maximum number of response items this stream will buffer before applying
    /// sender backpressure.
    #[must_use]
    pub fn buffer_capacity(&self) -> usize {
        MAX_STREAM_BUFFERED
    }

    /// Remaining enqueue slots before the stream applies sender backpressure.
    #[must_use]
    pub fn remaining_capacity(&self) -> usize {
        let len = lock_unpoisoned(&self.state).items.len();
        MAX_STREAM_BUFFERED.saturating_sub(len)
    }

    /// Whether the stream has reached its bounded buffer cap.
    #[must_use]
    pub fn is_full(&self) -> bool {
        lock_unpoisoned(&self.state).items.len() >= MAX_STREAM_BUFFERED
    }

    /// Polls for outbound buffer capacity, *suspending* the producer at the
    /// bounded-buffer window when it is exhausted.
    ///
    /// This is the suspend/resume counterpart to [`push`](Self::push): where
    /// `push` fails fast with
    /// [`Code::ResourceExhausted`](crate::grpc::status::Code::ResourceExhausted)
    /// once the buffer is full, `poll_reserve` instead parks the producer until
    /// the consumer drains an item (the "window update"), then wakes it so the
    /// next `push` is guaranteed a free slot. A server-streaming handler that
    /// gates each `push` behind `poll_reserve` is flow-controlled: it cannot
    /// outrun a slow consumer, and the buffer never exceeds
    /// [`MAX_STREAM_BUFFERED`].
    ///
    /// Returns:
    /// - `Poll::Ready(Ok(()))` when at least one slot is free,
    /// - `Poll::Pending` (registering `cx`'s waker) when the buffer is full,
    /// - `Poll::Ready(Err(_))` when the stream is closed — the producer should
    ///   stop rather than spin.
    pub fn poll_reserve(&self, cx: &mut Context<'_>) -> Poll<Result<(), Status>> {
        let mut state = lock_unpoisoned(&self.state);
        if state.closed {
            return Poll::Ready(Err(Status::failed_precondition(
                "cannot reserve capacity on a closed response stream",
            )));
        }
        if state.items.len() < MAX_STREAM_BUFFERED {
            return Poll::Ready(Ok(()));
        }
        state.register_producer_waiter(cx.waker());
        Poll::Pending
    }

    /// Close the stream.
    pub fn close(&self) {
        let (waiters, producers) = {
            let mut state = lock_unpoisoned(&self.state);
            state.closed = true;
            (state.take_waiters(), state.take_producer_waiters())
        };
        for waker in waiters {
            waker.wake();
        }
        // Wake parked producers so they re-poll, observe the close, and stop.
        for waker in producers {
            waker.wake();
        }
    }

    /// Close the stream with a terminal status.
    pub fn cancel(&self, status: Status) {
        self.cancel_with_metadata(status, Metadata::new());
    }

    fn set_terminal_status(
        &self,
        status: Status,
        metadata: Metadata,
        discard_buffered: bool,
    ) -> bool {
        let (waiters, producers, inserted) = {
            let mut state = lock_unpoisoned(&self.state);
            state.closed = true;
            let inserted = state.terminal_status.is_none();
            if state.terminal_status.is_none() {
                if discard_buffered {
                    state.items.clear();
                }
                state.terminal_status = Some(status);
                state.terminal_metadata = metadata;
            }
            (
                state.take_waiters(),
                state.take_producer_waiters(),
                inserted,
            )
        };
        for waker in waiters {
            waker.wake();
        }
        // Wake parked producers so they re-poll, observe the close, and stop.
        for waker in producers {
            waker.wake();
        }
        inserted
    }

    /// Cancel the stream immediately with a terminal status and trailing metadata.
    ///
    /// Cancellation is abrupt: queued response items are discarded so the
    /// caller observes the terminal status before any stale buffered payloads.
    pub fn cancel_with_metadata(&self, status: Status, metadata: Metadata) {
        if self.set_terminal_status(status, metadata, true)
            && let Some(on_cancel) = &self.on_cancel
        {
            let mut hook = lock_unpoisoned(on_cancel);
            let _ = hook();
        }
    }

    /// Finish the stream with a terminal status after draining queued items.
    ///
    /// This models the gRPC trailers path where already-received response data
    /// remains visible before the final status/trailers are observed.
    pub fn finish_with_metadata(&self, status: Status, metadata: Metadata) {
        self.set_terminal_status(status, metadata, false);
    }

    /// Returns the terminal trailing metadata captured for the stream.
    #[must_use]
    pub fn terminal_metadata(&self) -> Metadata {
        lock_unpoisoned(&self.state).terminal_metadata.clone()
    }
}

impl<T> Default for ResponseStream<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T: Send> Streaming for ResponseStream<T> {
    type Message = T;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Message, Status>>> {
        let mut state = lock_unpoisoned(&self.state);
        if let Some(item) = state.items.pop_front() {
            // A buffer slot just freed: wake any producer parked on
            // `poll_reserve` so it observes the window update and resumes.
            let producers = state.take_producer_waiters();
            drop(state);
            for waker in producers {
                waker.wake();
            }
            return Poll::Ready(Some(item));
        }
        if let Some(status) = state.terminal_status.take() {
            return Poll::Ready(Some(Err(status)));
        }
        if state.closed {
            return Poll::Ready(None);
        }
        state.register_waiter(cx.waker());
        Poll::Pending
    }
}

type SendHook<T> = Box<dyn FnMut(T) -> Result<(), Status> + Send>;
type CloseHook = Box<dyn FnMut() -> Result<(), Status> + Send>;

#[derive(Debug, Clone, Default)]
enum RequestSinkCloseState {
    #[default]
    Open,
    Graceful,
    Cancelled(Status),
    Failed(Status),
}

impl RequestSinkCloseState {
    fn is_open(&self) -> bool {
        matches!(self, Self::Open)
    }
}

#[derive(Default)]
struct RequestSinkState {
    close_state: RequestSinkCloseState,
    sent_count: usize,
    last_message: Option<Box<dyn Any + Send>>,
    waiter: Option<Waker>,
}

impl RequestSinkState {
    fn new() -> Self {
        Self::default()
    }
}

/// A sink for sending requests to the server.
pub struct RequestSink<T> {
    state: Arc<Mutex<RequestSinkState>>,
    on_send: Option<SendHook<T>>,
    on_close: Option<CloseHook>,
    on_cancel: Option<CloseHook>,
}

fn send_closed_error(close_state: &RequestSinkCloseState) -> Option<Status> {
    match close_state {
        RequestSinkCloseState::Open => None,
        RequestSinkCloseState::Graceful => Some(Status::failed_precondition(
            "cannot send after request sink is closed",
        )),
        RequestSinkCloseState::Cancelled(status) | RequestSinkCloseState::Failed(status) => {
            Some(status.clone())
        }
    }
}

fn cancel_request_sink_state(state: &Arc<Mutex<RequestSinkState>>, status: Status) {
    let waiter = {
        let mut state = lock_unpoisoned(state);
        if !state.close_state.is_open() {
            None
        } else {
            state.close_state = RequestSinkCloseState::Cancelled(status);
            state.waiter.take()
        }
    };
    if let Some(waiter) = waiter {
        waiter.wake();
    }
}

impl<T> RequestSink<T> {
    /// Create a new request sink.
    #[must_use]
    pub fn new() -> Self {
        Self {
            state: Arc::new(Mutex::new(RequestSinkState::new())),
            on_send: None,
            on_close: None,
            on_cancel: None,
        }
    }

    /// Return the number of request messages accepted by this sink.
    #[must_use]
    pub fn sent_count(&self) -> usize {
        lock_unpoisoned(&self.state).sent_count
    }

    fn from_state(state: Arc<Mutex<RequestSinkState>>) -> Self {
        Self {
            state,
            on_send: None,
            on_close: None,
            on_cancel: None,
        }
    }

    #[cfg(test)]
    fn with_hooks(
        on_send: Option<SendHook<T>>,
        on_close: Option<CloseHook>,
        on_cancel: Option<CloseHook>,
    ) -> Self {
        Self::with_state_and_hooks(
            Arc::new(Mutex::new(RequestSinkState::new())),
            on_send,
            on_close,
            on_cancel,
        )
    }

    fn with_state_and_hooks(
        state: Arc<Mutex<RequestSinkState>>,
        on_send: Option<SendHook<T>>,
        on_close: Option<CloseHook>,
        on_cancel: Option<CloseHook>,
    ) -> Self {
        Self {
            state,
            on_send,
            on_close,
            on_cancel,
        }
    }

    /// Send a request message.
    #[allow(clippy::unused_async)]
    pub async fn send(&mut self, message: T) -> Result<(), Status>
    where
        T: Send + 'static,
    {
        if self.on_send.is_none() {
            let closed_error = {
                let mut state = self
                    .state
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                let closed_error = send_closed_error(&state.close_state);
                if closed_error.is_none() {
                    if state.sent_count > 0 {
                        return Err(Status::failed_precondition(
                            "loopback client streaming does not support multiple request messages yet",
                        ));
                    }
                    state.last_message = Some(Box::new(message));
                    state.sent_count = state.sent_count.saturating_add(1);
                }
                drop(state);
                closed_error
            };
            if let Some(status) = closed_error {
                return Err(status);
            }
            return Ok(());
        }

        let closed_error = {
            let state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            send_closed_error(&state.close_state)
        };
        if let Some(status) = closed_error {
            return Err(status);
        }
        if let Some(hook) = self.on_send.as_mut() {
            hook(message)?;
        }
        {
            let mut state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            state.sent_count = state.sent_count.saturating_add(1);
        }
        Ok(())
    }

    /// Close the sink, signaling no more requests.
    #[allow(clippy::unused_async)]
    pub async fn close(&mut self) -> Result<(), Status> {
        {
            let state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            match &state.close_state {
                RequestSinkCloseState::Open => {}
                RequestSinkCloseState::Graceful => return Ok(()),
                RequestSinkCloseState::Cancelled(status)
                | RequestSinkCloseState::Failed(status) => {
                    return Err(status.clone());
                }
            }
        }
        if let Some(hook) = self.on_close.as_mut() {
            if let Err(status) = hook() {
                let waiter = {
                    let mut state = self
                        .state
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner);
                    state.close_state = RequestSinkCloseState::Failed(status.clone());
                    state.waiter.take()
                };
                if let Some(waiter) = waiter {
                    waiter.wake();
                }
                return Err(status);
            }
        }
        let waiter = {
            let mut state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            state.close_state = RequestSinkCloseState::Graceful;
            state.waiter.take()
        };
        if let Some(waiter) = waiter {
            waiter.wake();
        }
        Ok(())
    }
}

impl<T> Drop for RequestSink<T> {
    fn drop(&mut self) {
        let cancel_status = Status::cancelled("request stream cancelled by client");
        let (waiter, invoke_cancel_hook, invoke_close_hook) = {
            let mut state = lock_unpoisoned(&self.state);
            if !state.close_state.is_open() {
                (None, false, false)
            } else {
                state.close_state = RequestSinkCloseState::Cancelled(cancel_status);
                (
                    state.waiter.take(),
                    self.on_cancel.is_some(),
                    self.on_close.is_some(),
                )
            }
        };

        if let Some(waiter) = waiter {
            waiter.wake();
        }

        if invoke_cancel_hook {
            if let Some(hook) = self.on_cancel.as_mut() {
                let _ = hook();
            }
        } else if invoke_close_hook {
            if let Some(hook) = self.on_close.as_mut() {
                let _ = hook();
            }
        }
    }
}

impl<T> fmt::Debug for RequestSink<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        f.debug_struct("RequestSink")
            .field("close_state", &state.close_state)
            .field("sent_count", &state.sent_count)
            .field("has_send_hook", &self.on_send.is_some())
            .field("has_close_hook", &self.on_close.is_some())
            .field("has_cancel_hook", &self.on_cancel.is_some())
            .finish()
    }
}

impl<T> Default for RequestSink<T> {
    fn default() -> Self {
        Self::new()
    }
}

/// A future that resolves to a response.
pub struct ResponseFuture<T> {
    state: Arc<Mutex<RequestSinkState>>,
    resolver: Option<ResponseResolver<T>>,
}

type ResponseResolver<T> =
    Box<dyn FnMut(&mut RequestSinkState) -> Result<Response<T>, Status> + Send>;

impl<T> fmt::Debug for ResponseFuture<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        f.debug_struct("ResponseFuture")
            .field("sink_close_state", &state.close_state)
            .field("sink_sent_count", &state.sent_count)
            .field("has_resolver", &self.resolver.is_some())
            .finish()
    }
}

impl<T> ResponseFuture<T> {
    /// Create a new response future.
    #[must_use]
    pub fn new() -> Self {
        Self {
            state: Arc::new(Mutex::new(RequestSinkState {
                close_state: RequestSinkCloseState::Graceful,
                ..RequestSinkState::new()
            })),
            resolver: Some(Box::new(|_| {
                Err(Status::failed_precondition(
                    "response future is not linked to a request sink",
                ))
            })),
        }
    }

    fn with_resolver<F>(state: Arc<Mutex<RequestSinkState>>, resolver: F) -> Self
    where
        F: FnMut(&mut RequestSinkState) -> Result<Response<T>, Status> + Send + 'static,
    {
        Self {
            state,
            resolver: Some(Box::new(resolver)),
        }
    }
}

impl<T> Default for ResponseFuture<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T: Send> Future for ResponseFuture<T> {
    type Output = Result<Response<T>, Status>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let mut state = lock_unpoisoned(&this.state);
        if state.close_state.is_open() {
            if !state
                .waiter
                .as_ref()
                .is_some_and(|w| w.will_wake(cx.waker()))
            {
                state.waiter = Some(cx.waker().clone());
            }
            drop(state);
            return Poll::Pending;
        }
        let Some(mut resolver) = this.resolver.take() else {
            drop(state);
            return Poll::Ready(Err(Status::failed_precondition(
                "response future has already completed",
            )));
        };
        let output = match state.close_state.clone() {
            RequestSinkCloseState::Graceful => resolver(&mut state),
            RequestSinkCloseState::Cancelled(status) | RequestSinkCloseState::Failed(status) => {
                Err(status)
            }
            RequestSinkCloseState::Open => unreachable!("open sinks must have returned Pending"),
        };
        drop(state);
        Poll::Ready(output)
    }
}

/// Client interceptor for modifying requests.
pub trait ClientInterceptor: Send + Sync {
    /// Intercept a request before it is sent.
    fn intercept(&self, request: &mut Request<Bytes>) -> Result<(), Status>;
}

impl<T> ClientInterceptor for T
where
    T: super::server::Interceptor,
{
    fn intercept(&self, request: &mut Request<Bytes>) -> Result<(), Status> {
        self.intercept_request(request)
    }
}

/// A client interceptor that adds metadata to requests.
#[derive(Debug, Clone)]
pub struct MetadataInterceptor {
    /// Metadata to add.
    metadata: Metadata,
}

impl MetadataInterceptor {
    /// Create a new metadata interceptor.
    #[must_use]
    pub fn new() -> Self {
        Self {
            metadata: Metadata::new(),
        }
    }

    /// Add an ASCII metadata value.
    #[must_use]
    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        let _ = self.metadata.insert(key, value);
        self
    }
}

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

impl ClientInterceptor for MetadataInterceptor {
    fn intercept(&self, request: &mut Request<Bytes>) -> Result<(), Status> {
        let request_metadata = request.metadata_mut();
        request_metadata.reserve(self.metadata.len());
        for (key, value) in self.metadata.iter() {
            match value {
                super::streaming::MetadataValue::Ascii(v) => {
                    let _ = request_metadata.insert(key, v.clone());
                }
                super::streaming::MetadataValue::Binary(v) => {
                    let _ = request_metadata.insert_bin(key, v.clone());
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    #![allow(
        clippy::pedantic,
        clippy::nursery,
        clippy::expect_fun_call,
        clippy::map_unwrap_or,
        clippy::cast_possible_wrap,
        clippy::future_not_send,
        unused_must_use
    )]
    use super::*;
    use crate::codec::Encoder;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::task::{Context, Poll, Wake, Waker};

    struct CountWaker(Arc<AtomicUsize>);

    impl Wake for CountWaker {
        fn wake(self: Arc<Self>) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }

        fn wake_by_ref(self: &Arc<Self>) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }

    fn counting_waker(counter: &Arc<AtomicUsize>) -> Waker {
        Waker::from(Arc::new(CountWaker(Arc::clone(counter))))
    }

    fn poll_stream<T: Send>(
        stream: &mut ResponseStream<T>,
        waker: &Waker,
    ) -> Poll<Option<Result<T, Status>>> {
        let mut cx = Context::from_waker(waker);
        Streaming::poll_next(Pin::new(stream), &mut cx)
    }

    fn init_test(name: &str) {
        crate::test_utils::init_test_logging();
        crate::test_phase!(name);
    }

    #[test]
    fn test_channel_builder() {
        init_test("test_channel_builder");
        let builder = Channel::builder("http://loopback:50051")
            .connect_timeout(Duration::from_secs(10))
            .timeout(Duration::from_secs(30))
            .max_recv_message_size(8 * 1024 * 1024);

        crate::assert_with_log!(
            builder.config.connect_timeout == Duration::from_secs(10),
            "connect_timeout",
            Duration::from_secs(10),
            builder.config.connect_timeout
        );
        crate::assert_with_log!(
            builder.config.timeout == Some(Duration::from_secs(30)),
            "timeout",
            Some(Duration::from_secs(30)),
            builder.config.timeout
        );
        crate::assert_with_log!(
            builder.config.max_recv_message_size == 8 * 1024 * 1024,
            "max_recv_message_size",
            8 * 1024 * 1024,
            builder.config.max_recv_message_size
        );
        crate::test_complete!("test_channel_builder");
    }

    #[test]
    fn test_channel_config_default() {
        init_test("test_channel_config_default");
        let config = ChannelConfig::default();
        crate::assert_with_log!(
            config.connect_timeout == Duration::from_secs(5),
            "connect_timeout",
            Duration::from_secs(5),
            config.connect_timeout
        );
        let timeout_none = config.timeout.is_none();
        crate::assert_with_log!(timeout_none, "timeout none", true, timeout_none);
        crate::assert_with_log!(!config.use_tls, "use_tls", false, config.use_tls);
        crate::assert_with_log!(
            config.send_compression.is_none(),
            "send compression default",
            true,
            config.send_compression.is_none()
        );
        crate::assert_with_log!(
            config.accept_compression == vec![CompressionEncoding::Identity],
            "accept compression default",
            vec![CompressionEncoding::Identity],
            config.accept_compression
        );
        crate::test_complete!("test_channel_config_default");
    }

    #[test]
    fn test_metadata_interceptor() {
        init_test("test_metadata_interceptor");
        let interceptor = MetadataInterceptor::new()
            .with_metadata("x-custom-header", "value")
            .with_metadata("x-another", "value2");

        let mut request = Request::new(Bytes::new());
        interceptor.intercept(&mut request).unwrap();

        let has_custom = request.metadata().get("x-custom-header").is_some();
        crate::assert_with_log!(has_custom, "custom header", true, has_custom);
        let has_another = request.metadata().get("x-another").is_some();
        crate::assert_with_log!(has_another, "another header", true, has_another);
        crate::test_complete!("test_metadata_interceptor");
    }

    // Pure data-type tests (wave 14 – CyanBarn)

    #[test]
    fn channel_config_debug_clone() {
        let cfg = ChannelConfig::default();
        let dbg = format!("{cfg:?}");
        assert!(dbg.contains("ChannelConfig"));

        let cloned = cfg;
        assert_eq!(cloned.connect_timeout, Duration::from_secs(5));
    }

    #[test]
    fn channel_config_default_values() {
        let cfg = ChannelConfig::default();
        assert_eq!(cfg.connect_timeout, Duration::from_secs(5));
        assert!(cfg.timeout.is_none());
        assert_eq!(cfg.max_recv_message_size, 4 * 1024 * 1024);
        assert_eq!(cfg.max_send_message_size, 4 * 1024 * 1024);
        assert_eq!(cfg.initial_connection_window_size, 1024 * 1024);
        assert_eq!(cfg.initial_stream_window_size, 1024 * 1024);
        assert!(cfg.keepalive_interval.is_none());
        assert!(cfg.keepalive_timeout.is_none());
        assert!(!cfg.use_tls);
        assert!(cfg.send_compression.is_none());
        assert_eq!(cfg.accept_compression, vec![CompressionEncoding::Identity]);
    }

    #[test]
    fn channel_builder_debug() {
        let builder = Channel::builder("http://loopback:50051");
        let dbg = format!("{builder:?}");
        assert!(dbg.contains("ChannelBuilder"));
        assert!(dbg.contains("loopback"));
    }

    #[test]
    fn channel_builder_all_setters() {
        let builder = Channel::builder("http://host:443")
            .connect_timeout(Duration::from_secs(30))
            .timeout(Duration::from_secs(60))
            .max_recv_message_size(1024)
            .max_send_message_size(2048)
            .initial_connection_window_size(512)
            .initial_stream_window_size(256)
            .keepalive_interval(Duration::from_secs(10))
            .keepalive_timeout(Duration::from_secs(5))
            .send_compression(CompressionEncoding::Gzip)
            .accept_compressions([CompressionEncoding::Identity, CompressionEncoding::Gzip])
            .tls();

        assert_eq!(builder.config.connect_timeout, Duration::from_secs(30));
        assert_eq!(builder.config.timeout, Some(Duration::from_secs(60)));
        assert_eq!(builder.config.max_recv_message_size, 1024);
        assert_eq!(builder.config.max_send_message_size, 2048);
        assert_eq!(builder.config.initial_connection_window_size, 512);
        assert_eq!(builder.config.initial_stream_window_size, 256);
        assert_eq!(
            builder.config.keepalive_interval,
            Some(Duration::from_secs(10))
        );
        assert_eq!(
            builder.config.keepalive_timeout,
            Some(Duration::from_secs(5))
        );
        assert_eq!(
            builder.config.send_compression,
            Some(CompressionEncoding::Gzip)
        );
        assert_eq!(
            builder.config.accept_compression,
            vec![CompressionEncoding::Identity, CompressionEncoding::Gzip]
        );
        assert!(builder.config.use_tls);
    }

    fn make_channel(uri: &str) -> Channel {
        futures_lite::future::block_on(Channel::connect(uri)).unwrap()
    }

    #[test]
    fn channel_debug_clone() {
        let channel = make_channel("http://loopback:8080");
        let dbg = format!("{channel:?}");
        assert!(dbg.contains("Channel"));

        let cloned = channel;
        assert_eq!(cloned.uri(), "http://loopback:8080");
    }

    #[test]
    fn channel_uri_accessor() {
        let channel = make_channel("http://loopback:9090");
        assert_eq!(channel.uri(), "http://loopback:9090");
        assert_eq!(channel.config().connect_timeout, Duration::from_secs(5));
    }

    #[test]
    fn grpc_client_debug() {
        let channel = make_channel("http://loopback:50051");
        let client = GrpcClient::new(channel);
        let dbg = format!("{client:?}");
        assert!(dbg.contains("GrpcClient"));
    }

    #[test]
    fn grpc_client_channel_accessor() {
        let channel = make_channel("http://loopback:80");
        let client = GrpcClient::new(channel);
        assert_eq!(client.channel().uri(), "http://loopback:80");
    }

    #[test]
    fn grpc_client_applies_deadline_metadata_by_default() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .timeout(Duration::from_secs(2))
                .connect(),
        )
        .expect("channel");
        let mut client = GrpcClient::new(channel);
        let response: Response<String> = futures_lite::future::block_on(
            client.unary("/pkg.Service/Method", Request::new("hello".to_owned())),
        )
        .expect("unary");

        match response.metadata().get("grpc-timeout") {
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "2S");
            }
            other => panic!("expected grpc-timeout metadata, got: {other:?}"),
        }
    }

    #[test]
    fn grpc_client_repairs_malformed_timeout_before_building_outbound_metadata() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .timeout(Duration::from_secs(2))
                .connect(),
        )
        .expect("channel");
        let client = GrpcClient::new(channel);
        let mut request = Request::new(Bytes::new());
        request.metadata_mut().insert("grpc-timeout", "bogus");

        let metadata = client
            .build_outbound_metadata(&request, "/pkg.Service/Method")
            .expect("metadata");

        match metadata.get("grpc-timeout") {
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "2S");
            }
            other => panic!("expected repaired grpc-timeout metadata, got: {other:?}"),
        }

        let timeout_count = metadata
            .iter()
            .filter(|(key, _)| key.eq_ignore_ascii_case("grpc-timeout"))
            .count();
        assert_eq!(timeout_count, 1);
    }

    /// br-asupersync-20occs: when channel.config.timeout is None AND the
    /// request metadata carries a malformed grpc-timeout, the malformed
    /// entry must be SCRUBBED before send. Previously the value rode
    /// through to the wire because the existing-state classification only
    /// distinguished parseable-or-fall-back-to-default, and with no
    /// default to insert, the malformed entry was left untouched.
    #[test]
    fn occs20_malformed_grpc_timeout_scrubbed_when_channel_timeout_is_none() {
        let channel = futures_lite::future::block_on(
            // No .timeout(...) — channel.config.timeout is None.
            Channel::builder("http://loopback:80").connect(),
        )
        .expect("channel");
        let client = GrpcClient::new(channel);
        let mut request = Request::new(Bytes::new());
        request.metadata_mut().insert("grpc-timeout", "bogus");

        let metadata = client
            .build_outbound_metadata(&request, "/pkg.Service/Method")
            .expect("metadata");

        // Malformed entry MUST be scrubbed; with no channel default and no
        // valid request value, no grpc-timeout entry should remain.
        assert!(
            metadata.get("grpc-timeout").is_none(),
            "malformed grpc-timeout must be scrubbed when channel timeout is None, got: {:?}",
            metadata.get("grpc-timeout")
        );
        let timeout_count = metadata
            .iter()
            .filter(|(key, _)| key.eq_ignore_ascii_case("grpc-timeout"))
            .count();
        assert_eq!(
            timeout_count, 0,
            "no grpc-timeout entries should remain after scrub"
        );
    }

    /// br-asupersync-20occs: positive control — well-formed grpc-timeout
    /// passes through unchanged regardless of channel default.
    #[test]
    fn occs20_well_formed_grpc_timeout_passes_through_with_no_channel_default() {
        let channel =
            futures_lite::future::block_on(Channel::builder("http://loopback:80").connect())
                .expect("channel");
        let client = GrpcClient::new(channel);
        let mut request = Request::new(Bytes::new());
        request.metadata_mut().insert("grpc-timeout", "100m");

        let metadata = client
            .build_outbound_metadata(&request, "/pkg.Service/Method")
            .expect("metadata");

        match metadata.get("grpc-timeout") {
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "100m");
            }
            other => panic!("expected preserved grpc-timeout, got: {other:?}"),
        }
    }

    #[test]
    fn grpc_client_interceptors_and_compression_metadata_are_applied() {
        use crate::grpc::timeout_interceptor;

        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .send_compression(CompressionEncoding::Gzip)
                .accept_compressions([CompressionEncoding::Identity, CompressionEncoding::Gzip])
                .connect(),
        )
        .expect("channel");

        let mut client = GrpcClient::new(channel)
            .with_interceptor(timeout_interceptor(777))
            .with_interceptor(MetadataInterceptor::new().with_metadata("x-client-id", "cobalt"));

        let response: Response<String> = futures_lite::future::block_on(
            client.unary("/pkg.Service/Method", Request::new("hello".to_owned())),
        )
        .expect("unary");

        let metadata = response.metadata();
        match metadata.get("grpc-timeout") {
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "777m");
            }
            other => panic!("expected interceptor timeout metadata, got: {other:?}"),
        }
        match metadata.get("grpc-encoding") {
            #[cfg(feature = "compression")]
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "gzip");
            }
            #[cfg(not(feature = "compression"))]
            None => {}
            other => panic!("unexpected grpc-encoding metadata: {other:?}"),
        }
        match metadata.get("grpc-accept-encoding") {
            #[cfg(feature = "compression")]
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "identity,gzip");
            }
            #[cfg(not(feature = "compression"))]
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "identity");
            }
            other => panic!("unexpected grpc-accept-encoding metadata: {other:?}"),
        }
        match metadata.get("x-client-id") {
            Some(super::super::streaming::MetadataValue::Ascii(value)) => {
                assert_eq!(value, "cobalt");
            }
            other => panic!("expected interceptor metadata, got: {other:?}"),
        }
    }

    #[test]
    fn grpc_client_identity_send_compression_keeps_uncompressed_frames() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .send_compression(CompressionEncoding::Identity)
                .connect(),
        )
        .expect("channel");

        let mut client = GrpcClient::new(channel);
        let mut framed = crate::bytes::BytesMut::new();
        client
            .codec
            .encode_message(&Bytes::from_static(b"hello"), &mut framed)
            .expect("identity framing must encode");

        assert_eq!(
            framed[0], 0,
            "identity send compression must not set compressed flag"
        );
        let buf = framed
            .split_off(crate::grpc::codec::MESSAGE_HEADER_SIZE)
            .freeze();
        assert_eq!(buf.as_ref(), b"hello");
    }

    #[test]
    #[cfg(not(feature = "compression"))]
    fn grpc_client_unsupported_gzip_send_compression_stays_uncompressed() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .send_compression(CompressionEncoding::Gzip)
                .accept_compression(CompressionEncoding::Gzip)
                .connect(),
        )
        .expect("channel");

        let mut client = GrpcClient::new(channel);
        let mut framed = crate::bytes::BytesMut::new();
        client
            .codec
            .encode_message(&Bytes::from_static(b"hello"), &mut framed)
            .expect("unsupported gzip must fall back to uncompressed framing");

        assert_eq!(
            framed[0], 0,
            "unsupported gzip config must not set compressed flag"
        );
    }

    #[test]
    #[cfg(feature = "compression")]
    fn grpc_client_gzip_send_compression_uses_gzip_frames() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .send_compression(CompressionEncoding::Gzip)
                .accept_compression(CompressionEncoding::Gzip)
                .connect(),
        )
        .expect("channel");

        let mut client = GrpcClient::new(channel);
        let mut framed = crate::bytes::BytesMut::new();
        client
            .codec
            .encode_message(&Bytes::from_static(b"hello gzip"), &mut framed)
            .expect("gzip framing must encode");

        assert_eq!(
            framed[0], 1,
            "gzip send compression must set compressed flag"
        );
        assert_eq!(
            &framed[crate::grpc::codec::MESSAGE_HEADER_SIZE
                ..crate::grpc::codec::MESSAGE_HEADER_SIZE + 2],
            &[0x1f, 0x8b]
        );
    }

    #[test]
    fn grpc_client_codec_applies_channel_message_limits() {
        let channel = futures_lite::future::block_on(
            Channel::builder("http://loopback:80")
                .max_send_message_size(3)
                .max_recv_message_size(5)
                .connect(),
        )
        .expect("channel");

        let mut client = GrpcClient::new(channel);

        let encode_err = client
            .codec
            .encode_message(
                &Bytes::from_static(b"abcd"),
                &mut crate::bytes::BytesMut::new(),
            )
            .expect_err("send limit should be applied to the live client codec");
        assert!(matches!(encode_err, GrpcError::MessageTooLarge));

        let mut encoded = crate::bytes::BytesMut::new();
        let mut framing = crate::grpc::codec::GrpcCodec::new();
        framing
            .encode(
                crate::grpc::codec::GrpcMessage::new(Bytes::from_static(b"123456")),
                &mut encoded,
            )
            .expect("producer encode must succeed");

        let decode_err = client
            .codec
            .decode_message(&mut encoded)
            .expect_err("recv limit should be applied to the live client codec");
        assert!(matches!(decode_err, GrpcError::MessageTooLarge));
    }

    #[test]
    fn encode_grpc_timeout_prefers_largest_unit_with_eight_digit_limit() {
        assert_eq!(encode_grpc_timeout(Duration::from_secs(2)), "2S");
        assert_eq!(encode_grpc_timeout(Duration::from_millis(1)), "1m");
        assert_eq!(encode_grpc_timeout(Duration::from_nanos(1)), "1n");
        assert_eq!(encode_grpc_timeout(Duration::from_micros(1500)), "1500u");
    }

    #[test]
    fn validate_rpc_path_rejects_empty_or_extra_segments() {
        for path in ["/test.Svc/", "//Method", "/test.Svc/Method/Extra"] {
            let status = validate_rpc_path(path).expect_err("path should be rejected");
            assert_eq!(status.code(), crate::grpc::Code::InvalidArgument);
        }
        assert!(validate_rpc_path("/test.Svc/Method").is_ok());
    }

    #[test]
    fn metadata_interceptor_debug() {
        let interceptor = MetadataInterceptor::new();
        let dbg = format!("{interceptor:?}");
        assert!(dbg.contains("MetadataInterceptor"));
    }

    #[test]
    fn metadata_interceptor_empty() {
        let interceptor = MetadataInterceptor::new();
        let mut request = Request::new(Bytes::new());
        interceptor.intercept(&mut request).unwrap();
        // No headers added - request should still have empty metadata
        assert!(request.metadata().get("nonexistent").is_none());
    }

    // Pure data-type tests (wave 34 – CyanBarn)

    #[test]
    fn response_stream_debug() {
        let stream = ResponseStream::<u8>::new();
        let dbg = format!("{stream:?}");
        assert!(dbg.contains("ResponseStream"));
    }

    #[test]
    fn response_stream_default() {
        let stream = ResponseStream::<i32>::default();
        let dbg = format!("{stream:?}");
        assert!(dbg.contains("ResponseStream"));
    }

    #[test]
    fn response_stream_supports_non_unpin_messages() {
        use std::marker::PhantomPinned;

        struct NonUnpin {
            _pin: PhantomPinned,
        }

        let mut stream = ResponseStream::open();
        stream
            .push(Ok(NonUnpin {
                _pin: PhantomPinned,
            }))
            .unwrap();
        stream.close();

        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(first.is_some());

        let second = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(second.is_none());
    }

    #[test]
    fn response_stream_push_rejects_when_buffer_full_and_recovers_after_drain() {
        init_test("response_stream_push_rejects_when_buffer_full_and_recovers_after_drain");
        let mut stream = ResponseStream::<u32>::open();
        for i in 0..MAX_STREAM_BUFFERED as u32 {
            stream.push(Ok(i)).expect("push before saturation succeeds");
        }

        let err = stream
            .push(Ok(MAX_STREAM_BUFFERED as u32))
            .expect_err("push past cap must fail");
        assert_eq!(err.code(), crate::grpc::Code::ResourceExhausted);

        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(matches!(first, Some(Ok(0))));

        stream
            .push(Ok(MAX_STREAM_BUFFERED as u32))
            .expect("push should succeed after draining one slot");
    }

    #[test]
    fn response_stream_clones_keep_all_pending_readers_wakeable() {
        let mut stream = ResponseStream::<u32>::open();
        let mut first_reader = stream.clone();
        let mut second_reader = stream.clone();
        let first_wake_count = Arc::new(AtomicUsize::new(0));
        let second_wake_count = Arc::new(AtomicUsize::new(0));
        let first_reader_waker = counting_waker(&first_wake_count);
        let second_reader_waker = counting_waker(&second_wake_count);

        assert!(poll_stream(&mut first_reader, &first_reader_waker).is_pending());
        assert!(poll_stream(&mut second_reader, &second_reader_waker).is_pending());

        stream
            .push(Ok(7))
            .expect("push should wake pending readers");
        assert_eq!(
            first_wake_count.load(Ordering::SeqCst),
            1,
            "first cloned reader lost its wakeup",
        );
        assert_eq!(
            second_wake_count.load(Ordering::SeqCst),
            1,
            "second cloned reader should also be notified",
        );

        assert!(matches!(
            poll_stream(&mut first_reader, &first_reader_waker),
            Poll::Ready(Some(Ok(7)))
        ));
        assert!(poll_stream(&mut second_reader, &second_reader_waker).is_pending());

        stream.close();
        assert_eq!(
            second_wake_count.load(Ordering::SeqCst),
            2,
            "close should wake the still-pending cloned reader",
        );
        assert!(matches!(
            poll_stream(&mut second_reader, &second_reader_waker),
            Poll::Ready(None)
        ));
    }

    #[test]
    fn response_stream_terminal_metadata_survives_terminal_error() {
        let mut stream = ResponseStream::<u32>::open();
        stream.push(Ok(7)).expect("data item should enqueue");

        let mut trailers = Metadata::new();
        assert!(trailers.insert_bin(
            "grpc-status-details-bin",
            Bytes::from_static(b"error-details"),
        ));
        trailers.insert("x-debug-trailer", "final-hop");
        stream.finish_with_metadata(Status::internal("stream failed"), trailers.clone());

        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(matches!(first, Some(Ok(7))));

        let second = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        match second {
            Some(Err(status)) => {
                assert_eq!(status.code(), crate::grpc::Code::Internal);
                assert_eq!(status.message(), "stream failed");
            }
            other => panic!("expected terminal status, got {other:?}"),
        }

        let stored = stream.terminal_metadata();
        assert!(matches!(
            stored.get("grpc-status-details-bin"),
            Some(crate::grpc::MetadataValue::Binary(value)) if value.as_ref() == b"error-details"
        ));
        assert!(matches!(
            stored.get("x-debug-trailer"),
            Some(crate::grpc::MetadataValue::Ascii(value)) if value == "final-hop"
        ));

        let third = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(third.is_none());
    }

    #[test]
    fn response_stream_cancel_discards_buffered_items_before_terminal_status() {
        let mut stream = ResponseStream::<u32>::open();
        stream.push(Ok(7)).expect("data item should enqueue");

        let mut trailers = Metadata::new();
        assert!(trailers.insert_bin("grpc-status-details-bin", Bytes::from_static(b"cancelled"),));
        stream.cancel_with_metadata(Status::cancelled("client cancelled stream"), trailers);

        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        match first {
            Some(Err(status)) => {
                assert_eq!(status.code(), crate::grpc::Code::Cancelled);
                assert_eq!(status.message(), "client cancelled stream");
            }
            other => panic!("expected immediate cancelled status, got {other:?}"),
        }

        let second = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(
            second.is_none(),
            "cancelled stream must terminate after status"
        );
    }

    #[test]
    fn request_sink_debug() {
        let sink = RequestSink::<u8>::new();
        let dbg = format!("{sink:?}");
        assert!(dbg.contains("RequestSink"));
    }

    #[test]
    fn request_sink_default() {
        let sink = RequestSink::<i32>::default();
        let dbg = format!("{sink:?}");
        assert!(dbg.contains("RequestSink"));
    }

    #[test]
    fn request_sink_close_hook_runs_once_when_closed_then_dropped() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let close_count = Arc::new(AtomicUsize::new(0));
        let hook_count = Arc::clone(&close_count);
        let mut sink: RequestSink<u32> = RequestSink::with_hooks(
            None,
            Some(Box::new(move || {
                hook_count.fetch_add(1, Ordering::SeqCst);
                Ok(())
            })),
            None,
        );

        futures_lite::future::block_on(sink.close()).expect("close should succeed");
        drop(sink);

        assert_eq!(
            close_count.load(Ordering::SeqCst),
            1,
            "close hook should run exactly once"
        );
    }

    #[test]
    fn request_sink_failed_send_hook_does_not_increment_sent_count() {
        let mut sink = RequestSink::with_hooks(
            Some(Box::new(|_: u32| {
                Err(Status::internal("send hook rejected the message"))
            })),
            None,
            None,
        );

        let error = futures_lite::future::block_on(sink.send(7))
            .expect_err("failing send hook must reject the message");
        assert_eq!(error.code(), crate::grpc::Code::Internal);

        {
            let state = lock_unpoisoned(&sink.state);
            assert_eq!(
                state.sent_count, 0,
                "failed sends must not be counted as successfully sent",
            );
            drop(state);
        }
    }

    #[test]
    fn request_sink_successful_send_hook_increments_sent_count() {
        let mut sink = RequestSink::with_hooks(Some(Box::new(|_: u32| Ok(()))), None, None);

        futures_lite::future::block_on(sink.send(7))
            .expect("successful send hook should accept the message");

        assert_eq!(
            lock_unpoisoned(&sink.state).sent_count,
            1,
            "successful sends must be counted"
        );
    }

    #[test]
    fn response_future_default() {
        let _fut = ResponseFuture::<i32>::default();
        // ResponseFuture does not derive Debug, but Default is implemented
    }

    #[test]
    fn response_future_new_fails_fast() {
        let response = futures_lite::future::block_on(ResponseFuture::<u8>::new())
            .expect_err("unlinked response future must fail immediately");
        assert_eq!(response.code(), crate::grpc::Code::FailedPrecondition);
    }

    #[test]
    fn metadata_interceptor_clone() {
        let interceptor = MetadataInterceptor::new().with_metadata("x-key", "val");
        let cloned = interceptor;
        let mut request = Request::new(Bytes::new());
        cloned.intercept(&mut request).unwrap();
        assert!(request.metadata().get("x-key").is_some());
    }

    #[test]
    fn metadata_interceptor_default() {
        let interceptor = MetadataInterceptor::default();
        let dbg = format!("{interceptor:?}");
        assert!(dbg.contains("MetadataInterceptor"));
    }

    #[test]
    fn client_streaming_future_resolves_when_sink_is_dropped() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (sink, future) = futures_lite::future::block_on(
            client.client_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("client streaming setup");

        // Dropping the sink should close the stream and wake the response future.
        drop(sink);
        let result = futures_lite::future::block_on(future);
        assert!(
            result.is_err(),
            "empty dropped stream should resolve with an error"
        );
    }

    #[test]
    fn bidi_stream_closes_when_sink_is_dropped() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (sink, mut stream) = futures_lite::future::block_on(
            client.bidi_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("bidi streaming setup");

        drop(sink);
        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        let status = first.expect("drop should surface a terminal status");
        assert_eq!(
            status
                .expect_err("drop should cancel bidi response stream")
                .code(),
            crate::grpc::Code::Cancelled
        );
        let second = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(second.is_none(), "cancelled bidi stream should then close");
    }

    /// GRPC-CONF-012: Client cancellation must surface CANCELLED immediately.
    /// Buffered loopback responses must not leak after the client aborts the RPC.
    #[test]
    fn conformance_bidi_stream_cancellation_suppresses_buffered_responses() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (mut sink, mut stream) = futures_lite::future::block_on(
            client.bidi_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("bidi streaming setup");

        futures_lite::future::block_on(sink.send(7))
            .expect("loopback bidi stream should buffer one echoed response");
        drop(sink);

        let first = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        match first {
            Some(Err(status)) => {
                assert_eq!(status.code(), crate::grpc::Code::Cancelled);
                assert_eq!(status.message(), "request stream cancelled by client");
            }
            other => panic!("expected immediate CANCELLED after client abort, got {other:?}"),
        }

        let second = futures_lite::future::block_on(futures_lite::future::poll_fn(|cx| {
            Streaming::poll_next(Pin::new(&mut stream), cx)
        }));
        assert!(second.is_none(), "cancelled bidi stream should then close");
    }

    #[test]
    fn conformance_bidi_response_cancel_closes_request_sink() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (mut sink, stream) = futures_lite::future::block_on(
            client.bidi_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("bidi streaming setup");

        stream.cancel(Status::cancelled("response stream cancelled by client"));

        let send_error = futures_lite::future::block_on(sink.send(7))
            .expect_err("response cancellation should cancel the request sink");
        assert_eq!(send_error.code(), crate::grpc::Code::Cancelled);
        assert_eq!(send_error.message(), "response stream cancelled by client");

        let close_error = futures_lite::future::block_on(sink.close())
            .expect_err("closing a cancelled request sink should report cancellation");
        assert_eq!(close_error.code(), crate::grpc::Code::Cancelled);
        assert_eq!(close_error.message(), "response stream cancelled by client");
    }

    #[test]
    fn client_streaming_drop_after_send_returns_cancelled() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (mut sink, future) = futures_lite::future::block_on(
            client.client_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("client streaming setup");

        futures_lite::future::block_on(sink.send(7)).expect("send should succeed");
        drop(sink);

        let error = futures_lite::future::block_on(future)
            .expect_err("dropped request stream must resolve as cancelled");
        assert_eq!(error.code(), crate::grpc::Code::Cancelled);
    }

    #[test]
    fn client_streaming_second_message_fails_closed() {
        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel);

        let (mut sink, future) = futures_lite::future::block_on(
            client.client_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect("client streaming setup");

        futures_lite::future::block_on(sink.send(7)).expect("first send should succeed");
        let error = futures_lite::future::block_on(sink.send(9))
            .expect_err("second send must fail closed in loopback mode");
        assert_eq!(error.code(), crate::grpc::Code::FailedPrecondition);

        futures_lite::future::block_on(sink.close()).expect("close should still succeed");
        let response =
            futures_lite::future::block_on(future).expect("first request should still resolve");
        assert_eq!(*response.get_ref(), 7);
    }

    #[test]
    fn request_sink_close_hook_failure_propagates_to_response_future() {
        let state = Arc::new(Mutex::new(RequestSinkState::new()));
        let mut sink: RequestSink<u32> = RequestSink {
            state: Arc::clone(&state),
            on_send: None,
            on_close: Some(Box::new(|| Err(Status::internal("close failed")))),
            on_cancel: None,
        };
        let future = ResponseFuture::with_resolver(state, |_| {
            Ok(Response::with_metadata(7_u32, Metadata::new()))
        });

        let close_error =
            futures_lite::future::block_on(sink.close()).expect_err("close hook should fail");
        assert_eq!(close_error.code(), crate::grpc::Code::Internal);

        let future_error = futures_lite::future::block_on(future)
            .expect_err("response future should reflect close failure");
        assert_eq!(future_error.code(), crate::grpc::Code::Internal);
    }

    #[test]
    fn bidi_streaming_applies_interceptors() {
        #[derive(Debug, Clone, Copy)]
        struct RejectInterceptor;

        impl crate::grpc::server::Interceptor for RejectInterceptor {
            fn intercept_request(&self, _request: &mut Request<Bytes>) -> Result<(), Status> {
                Err(Status::unauthenticated("blocked by interceptor"))
            }

            fn intercept_response(&self, _response: &mut Response<Bytes>) -> Result<(), Status> {
                Ok(())
            }
        }

        let channel = make_channel("http://loopback:50051");
        let mut client = GrpcClient::new(channel).with_interceptor(RejectInterceptor);

        let error = futures_lite::future::block_on(
            client.bidi_streaming::<u32, u32>("/pkg.Service/Method"),
        )
        .expect_err("bidi call should respect client interceptors");
        assert_eq!(error.code(), crate::grpc::Code::Unauthenticated);
    }

    #[test]
    fn channel_connect_accepts_loopback_and_localhost_hosts() {
        for uri in [
            "http://loopback:50051",
            "http://localhost:50051",
            "http://127.0.0.1:50051",
        ] {
            let channel = futures_lite::future::block_on(Channel::connect(uri))
                .expect("loopback and localhost targets should connect");
            assert_eq!(channel.uri(), uri);
        }
    }

    #[test]
    fn channel_connect_requires_explicit_tls_connector() {
        for uri in [
            "https://LOCALHOST:50051/service",
            "HTTPS://LOCALHOST:50051/service",
        ] {
            let error = futures_lite::future::block_on(Channel::connect(uri))
                .expect_err("HTTPS gRPC channel without a connector must fail closed");
            match error {
                GrpcError::Transport(TransportErrorKind::ProtocolViolation, message) => {
                    assert!(
                        message.contains("explicit TLS connector"),
                        "expected explicit-authority message for {uri}, got: {message}"
                    );
                }
                other => panic!(
                    "expected TLS protocol-violation transport error for {uri}, got: {other:?}"
                ),
            }
        }

        let error = futures_lite::future::block_on(
            Channel::builder("http://loopback:50051").tls().connect(),
        )
        .expect_err("deterministic loopback TLS must fail closed");
        assert!(error.to_string().contains("loopback channels"));
    }

    #[test]
    fn channel_connect_with_config_requires_explicit_tls_connector() {
        let config = ChannelConfig {
            use_tls: true,
            ..ChannelConfig::default()
        };
        let error = futures_lite::future::block_on(Channel::connect_with_config(
            "http://localhost:50051",
            config,
        ))
        .expect_err("TLS-enabled config must fail closed");

        match error {
            GrpcError::Transport(TransportErrorKind::ProtocolViolation, message) => {
                assert!(
                    message.contains("explicit TLS connector"),
                    "expected TLS enforcement message, got: {message}"
                );
            }
            other => panic!(
                "expected TLS protocol-violation transport error for direct config, got: {other:?}"
            ),
        }
    }

    #[test]
    fn channel_connect_rejects_non_localhost_host() {
        let error = futures_lite::future::block_on(Channel::connect("http://example.com:50051"))
            .expect_err("non-localhost target should fail closed");
        match error {
            GrpcError::Transport(_kind, message) => {
                assert!(message.contains("loopback and localhost only"));
            }
            other => panic!("expected transport error, got: {other:?}"),
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn explicit_dial_addr_requires_authenticated_native_https() {
        let address = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 50051));

        let cleartext = futures_lite::future::block_on(
            Channel::builder("http://grpc.service.invalid:50051")
                .dial_addr(address)
                .connect(),
        )
        .expect_err("explicit cleartext dial must fail closed");
        assert!(cleartext.to_string().contains("requires an https URI"));

        let missing_tls = futures_lite::future::block_on(
            Channel::builder("https://grpc.service.invalid:443")
                .dial_addr(address)
                .connect(),
        )
        .expect_err("explicit HTTPS dial without TLS authority must fail closed");
        assert!(missing_tls.to_string().contains("explicit TLS connector"));

        let deterministic_loopback = futures_lite::future::block_on(
            Channel::builder("https://loopback:443")
                .dial_addr(address)
                .connect(),
        )
        .expect_err("deterministic loopback must reject a native dial capability");
        assert!(
            deterministic_loopback
                .to_string()
                .contains("cannot use an explicit native dial address")
        );

        let userinfo = futures_lite::future::block_on(
            Channel::builder("https://user@grpc.service.invalid:443")
                .dial_addr(address)
                .connect(),
        )
        .expect_err("explicit dial must reject URI userinfo");
        assert!(userinfo.to_string().contains("userinfo is not supported"));

        for uri in [
            "https://grpc.service.invalid:443:444",
            "https://grpc.service.invalid:notaport",
            "https://[grpc.service.invalid:443",
        ] {
            let malformed =
                futures_lite::future::block_on(Channel::builder(uri).dial_addr(address).connect())
                    .expect_err("ambiguous explicit-dial authority must fail closed");
            assert!(
                malformed.to_string().contains("at most one port")
                    || malformed.to_string().contains("unsigned 16-bit integer"),
                "unexpected explicit-dial authority error for {uri}: {malformed}"
            );
        }
    }

    #[test]
    fn channel_connect_rejects_userinfo_bypass() {
        // Regression: "loopback" in userinfo must not fool the host check.
        // RFC 3986 §3.2: authority = [userinfo "@"] host [":" port]
        for uri in [
            "http://loopback:pw@evil.com:80",
            "http://loopback@evil.com",
            "http://user:loopback@attacker.io:443/path",
        ] {
            let error = futures_lite::future::block_on(Channel::connect(uri))
                .expect_err(&format!("userinfo bypass must fail: {uri}"));
            match error {
                GrpcError::Transport(_kind, msg) => {
                    assert!(
                        msg.contains("loopback and localhost only"),
                        "expected loopback/localhost-only error for {uri}, got: {msg}"
                    );
                }
                other => panic!("expected transport error for {uri}, got: {other:?}"),
            }
        }
    }

    #[test]
    fn channel_connect_rejects_authority_whitespace() {
        for uri in [
            "http:// localhost:50051",
            "http://localhost :50051",
            "http://localhost:50051 ",
            "http://local\thost:50051",
            "http://localhost\n:50051",
        ] {
            let error = match futures_lite::future::block_on(Channel::connect(uri)) {
                Ok(_) => panic!("authority whitespace must fail closed: {uri:?}"),
                Err(error) => error,
            };
            match error {
                GrpcError::Transport(_kind, message) => {
                    assert!(
                        message.contains("authority cannot contain whitespace"),
                        "expected authority whitespace error for {uri:?}, got: {message}"
                    );
                }
                other => panic!("expected transport error for {uri:?}, got: {other:?}"),
            }
        }
    }

    // --- br-asupersync-server-stack-hardening-eeexl1.1.3: outbound ---
    // --- deadline attach from the remaining ambient Cx budget       ---

    mod budget_forwarding {
        use super::*;
        use crate::trace::TraceBufferHandle;
        use crate::trace::event::{TraceData, TraceEventKind};
        use crate::types::Budget;

        /// Installs an ambient Cx whose budget deadline is `remaining`
        /// from now, runs `f`, and returns its result plus the trace
        /// buffer attached to the ambient context.
        fn with_ambient_budget<R>(
            remaining: Duration,
            f: impl FnOnce() -> R,
        ) -> (R, TraceBufferHandle) {
            let now = crate::time::wall_now();
            let budget = Budget::INFINITE.tightened_by_timeout(now, remaining);
            let cx = crate::cx::Cx::for_request_with_budget(budget);
            let trace = TraceBufferHandle::new(64);
            cx.set_trace_buffer(trace.clone());
            let _guard = crate::cx::Cx::set_current(Some(cx));
            (f(), trace)
        }

        fn outbound_timeout(client: &mut GrpcClient, path: &str) -> Option<Duration> {
            let response: Response<String> = futures_lite::future::block_on(
                client.unary(path, Request::new("hello".to_owned())),
            )
            .expect("unary");
            match response.metadata().get("grpc-timeout") {
                Some(crate::grpc::streaming::MetadataValue::Ascii(value)) => {
                    crate::grpc::server::parse_grpc_timeout(value)
                }
                _ => None,
            }
        }

        #[test]
        fn ambient_budget_caps_channel_default() {
            let channel = futures_lite::future::block_on(
                Channel::builder("http://loopback:80")
                    .timeout(Duration::from_secs(30))
                    .connect(),
            )
            .expect("channel");
            let mut client = GrpcClient::new(channel);
            let (sent, _trace) = with_ambient_budget(Duration::from_secs(1), || {
                outbound_timeout(&mut client, "/pkg.Service/Method")
            });
            let sent = sent.expect("grpc-timeout must be present");
            assert!(
                sent <= Duration::from_secs(1),
                "budget must cap the channel default, sent {sent:?}"
            );
            assert!(
                sent > Duration::from_millis(500),
                "cap should reflect the remaining budget, sent {sent:?}"
            );
        }

        #[test]
        fn ambient_budget_alone_becomes_wire_deadline() {
            let channel =
                futures_lite::future::block_on(Channel::builder("http://loopback:80").connect())
                    .expect("channel");
            let mut client = GrpcClient::new(channel);
            let (sent, _trace) = with_ambient_budget(Duration::from_millis(500), || {
                outbound_timeout(&mut client, "/pkg.Service/Method")
            });
            let sent = sent.expect("budget-derived grpc-timeout must be sent");
            assert!(
                sent <= Duration::from_millis(500),
                "wire deadline must not exceed the remaining budget, sent {sent:?}"
            );
            assert!(sent > Duration::from_millis(250), "sent {sent:?}");
        }

        #[test]
        fn tighter_per_request_override_passes_through_verbatim() {
            let channel =
                futures_lite::future::block_on(Channel::builder("http://loopback:80").connect())
                    .expect("channel");
            let mut client = GrpcClient::new(channel);
            let ((), _trace) = with_ambient_budget(Duration::from_secs(10), || {
                let mut request = Request::new("hello".to_owned());
                request.metadata_mut().insert("grpc-timeout", "100m");
                let response: Response<String> =
                    futures_lite::future::block_on(client.unary("/pkg.Service/Method", request))
                        .expect("unary");
                match response.metadata().get("grpc-timeout") {
                    Some(crate::grpc::streaming::MetadataValue::Ascii(value)) => {
                        assert_eq!(value, "100m", "tighter override must pass through verbatim");
                    }
                    other => panic!("expected grpc-timeout, got {other:?}"),
                }
            });
        }

        #[test]
        fn expired_ambient_budget_fails_fast() {
            let channel = futures_lite::future::block_on(
                Channel::builder("http://loopback:80")
                    .timeout(Duration::from_secs(30))
                    .connect(),
            )
            .expect("channel");
            let mut client = GrpcClient::new(channel);
            let (result, _trace) = with_ambient_budget(Duration::ZERO, || {
                futures_lite::future::block_on(client.unary::<String, String>(
                    "/pkg.Service/Method",
                    Request::new("hello".to_owned()),
                ))
            });
            let status = result.expect_err("expired budget must fail fast");
            assert_eq!(status.code(), crate::grpc::Code::DeadlineExceeded);
        }

        #[test]
        fn forwarded_budget_trace_event_emitted() {
            let channel = futures_lite::future::block_on(
                Channel::builder("http://loopback:80")
                    .timeout(Duration::from_secs(30))
                    .connect(),
            )
            .expect("channel");
            let mut client = GrpcClient::new(channel);
            let (_sent, trace) = with_ambient_budget(Duration::from_secs(1), || {
                outbound_timeout(&mut client, "/pkg.Service/Method")
            });
            let forwarded: Vec<String> = trace
                .snapshot()
                .iter()
                .filter(|e| e.kind == TraceEventKind::UserTrace)
                .filter_map(|e| match &e.data {
                    TraceData::Message(msg)
                        if msg.starts_with("client.budget_forwarded proto=grpc ") =>
                    {
                        Some(msg.clone())
                    }
                    _ => None,
                })
                .collect();
            assert_eq!(
                forwarded.len(),
                1,
                "exactly one forwarded event expected, got {forwarded:?}"
            );
            assert!(forwarded[0].contains("base=channel"));
            assert!(forwarded[0].contains("grpc_timeout="));
        }
    }
}