diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index cc6d1eb50e..69a5e0946d 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -114,8 +114,8 @@ tools/cc_utils/                      @DataDog/libdatadog-php
 tools/sidecar_mockgen/               @DataDog/libdatadog-php
 libdd-data-pipeline/src/otlp/                           @DataDog/apm-sdk-capabilities-rust
 libdd-data-pipeline/tests/test_trace_exporter_otlp_export.rs @DataDog/apm-sdk-capabilities-rust
+libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs @DataDog/apm-sdk-capabilities-rust
 libdd-trace-utils/src/otlp_encoder/                     @DataDog/apm-sdk-capabilities-rust
 datadog-sidecar/src/service/ffe_exposures_flusher.rs    @DataDog/libdatadog-php @DataDog/libdatadog-apm @DataDog/feature-flagging-and-experimentation-sdk
 datadog-sidecar/src/service/ffe_metrics_flusher.rs      @DataDog/libdatadog-php @DataDog/libdatadog-apm @DataDog/feature-flagging-and-experimentation-sdk
 .github/workflows/nix.yml                               @DataDog/nix-guild @DataDog/apm-common-components-core
-
diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs
index 6016c32dcb..3525ffd034 100644
--- a/libdd-data-pipeline-ffi/src/trace_exporter.rs
+++ b/libdd-data-pipeline-ffi/src/trace_exporter.rs
@@ -495,11 +495,12 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_shared_runtime(
     )
 }
 
-/// Enables OTLP HTTP/JSON export and sets the endpoint URL.
+/// Enables OTLP trace export and sets the endpoint URL.
 ///
-/// When set, traces are sent to this URL in OTLP HTTP/JSON format instead of the Datadog
-/// agent. The host language is responsible for resolving the endpoint from its configuration
-/// (e.g. `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this function.
+/// When set, traces are sent to this URL using the protocol selected by
+/// `ddog_trace_exporter_config_set_otlp_protocol` instead of the Datadog agent. The host language
+/// is responsible for resolving the endpoint from its configuration (e.g.
+/// `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this function.
 #[no_mangle]
 pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_endpoint(
     config: Option<&mut TraceExporterConfig>,
@@ -519,9 +520,10 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_endpoint(
     )
 }
 
-/// Sets the OTLP export protocol. Accepts the OTel-standard values `http/json` (default) or
-/// `http/protobuf`; `grpc` is rejected as not yet supported. The host language resolves the value
+/// Sets the OTLP export protocol. Accepts the OTel-standard values `http/json` (default),
+/// `http/protobuf`, or `grpc`; unknown values are rejected. The host language resolves the value
 /// (e.g. from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`).
+/// The `grpc` protocol currently supports plaintext `http://` endpoints only.
 ///
 /// Has no effect unless an OTLP endpoint is also configured via
 /// `ddog_trace_exporter_config_set_otlp_endpoint`; without one, traces are sent to the
@@ -540,9 +542,6 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol(
                 Ok(s) => s,
                 Err(e) => return Some(e),
             };
-            // `FromStr` is the single source of truth for string -> OtlpProtocol. It accepts only
-            // the supported HTTP encodings (`http/json`, `http/protobuf`); `grpc` and any unknown
-            // value are rejected with an error, so an unsupported protocol can never be stored.
             match value.parse::<OtlpProtocol>() {
                 Ok(p) => {
                     handle.otlp_protocol = Some(p);
@@ -754,9 +753,9 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_output_to_log(
 
 /// Create a new TraceExporter instance.
 ///
-/// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces to
-/// that endpoint in OTLP over HTTP — JSON or protobuf per the configured protocol — instead of
-/// to the Datadog agent. The same payload (e.g. MessagePack) is passed to
+/// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces using
+/// the configured `http/json`, `http/protobuf`, or `grpc` protocol instead of the Datadog agent.
+/// The same payload (e.g. MessagePack) is passed to
 /// `ddog_trace_exporter_send`; the library decodes and converts it to OTLP when OTLP is enabled.
 ///
 /// # Arguments
@@ -1626,14 +1625,16 @@ mod tests {
                 Some(OtlpProtocol::HttpProtobuf)
             );
 
-            // "grpc" → InvalidArgument
             let mut config = Some(TraceExporterConfig::default());
             let error = ddog_trace_exporter_config_set_otlp_protocol(
                 config.as_mut(),
                 CharSlice::from("grpc"),
             );
-            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
-            ddog_trace_exporter_error_free(error);
+            assert_eq!(error, None);
+            assert_eq!(
+                config.as_ref().unwrap().otlp_protocol,
+                Some(OtlpProtocol::Grpc)
+            );
 
             // Garbage value → InvalidArgument
             let mut config = Some(TraceExporterConfig::default());
@@ -1714,9 +1715,9 @@ mod tests {
     }
 
     #[test]
-    fn set_otlp_protocol_rejects_grpc_and_unknown() {
+    fn set_otlp_protocol_rejects_unknown() {
         let mut cfg = TraceExporterConfig::default();
-        for bad in ["grpc", "nonsense"] {
+        for bad in ["nonsense", "grcp"] {
             let err = unsafe {
                 ddog_trace_exporter_config_set_otlp_protocol(Some(&mut cfg), CharSlice::from(bad))
             };
diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml
index 69f12b8b14..58845dad7b 100644
--- a/libdd-data-pipeline/Cargo.toml
+++ b/libdd-data-pipeline/Cargo.toml
@@ -57,8 +57,10 @@ libdd-capabilities-impl = { version = "4.0.0", path = "../libdd-capabilities-imp
 # (hyper/tokio/socket2) does not build for wasm32, so the whole gRPC path is gated off.
 tonic = { version = "0.14", default-features = false }
 prost = "0.14.1"
+h2 = "0.4"
 hyper = { workspace = true, features = ["client", "http2"] }
 hyper-util = { workspace = true, features = ["tokio"] }
+rand = "0.8.5"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 getrandom = { version = "0.2", features = ["js"] }
@@ -87,7 +89,6 @@ libdd-trace-utils = { path = "../libdd-trace-utils", features = [
 ] }
 httpmock = "0.8.0-alpha.1"
 prost = "0.14.1"
-rand = "0.8.5"
 tempfile.workspace = true
 tokio = { version = "1.23", features = [
     "rt",
@@ -97,7 +98,6 @@ tokio = { version = "1.23", features = [
 duplicate = "2.0.1"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
-h2 = "0.4"
 zstd = { version = "0.13", default-features = false }
 
 [target.'cfg(target_arch = "wasm32")'.dev-dependencies]
diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs
index 39dd54a38e..a0ae1de342 100644
--- a/libdd-data-pipeline/src/otlp/config.rs
+++ b/libdd-data-pipeline/src/otlp/config.rs
@@ -6,12 +6,7 @@
 use http::HeaderMap;
 use std::time::Duration;
 
-/// OTLP trace export protocol — selects the HTTP body encoding and `Content-Type`.
-///
-/// Only the HTTP encodings libdatadog actually supports are representable. A `grpc` value (e.g.
-/// resolved from the OTel-default `OTEL_EXPORTER_OTLP_PROTOCOL`) is rejected by
-/// [`FromStr`](std::str::FromStr) rather than represented here, so an unsupported protocol can
-/// never be constructed and silently mishandled downstream.
+/// OTLP trace export protocol: selects the wire transport and body encoding.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub enum OtlpProtocol {
     /// HTTP with a JSON body (`Content-Type: application/json`). The default.
@@ -19,6 +14,8 @@ pub enum OtlpProtocol {
     HttpJson,
     /// HTTP with a protobuf body (`Content-Type: application/x-protobuf`).
     HttpProtobuf,
+    /// gRPC over HTTP/2.
+    Grpc,
 }
 
 impl std::str::FromStr for OtlpProtocol {
@@ -27,37 +24,31 @@ impl std::str::FromStr for OtlpProtocol {
         match s {
             "http/json" => Ok(OtlpProtocol::HttpJson),
             "http/protobuf" => Ok(OtlpProtocol::HttpProtobuf),
-            // gRPC is a valid OTLP protocol in the OTel spec but is not implemented in
-            // libdatadog. Reject it explicitly so callers get a clean error at the parse
-            // boundary, rather than constructing an unsupported value that has to be guarded
-            // against everywhere downstream.
-            "grpc" => Err("OTLP gRPC export is not supported".to_string()),
+            "grpc" => Ok(OtlpProtocol::Grpc),
             other => Err(format!("unknown OTLP protocol: {other}")),
         }
     }
 }
 
 impl OtlpProtocol {
-    /// The HTTP `Content-Type` for this protocol's body encoding. Crate-internal: the public type
-    /// is only constructed/selected by callers; encoding is the exporter's job.
-    pub(crate) fn content_type(&self) -> http::HeaderValue {
+    pub(crate) fn content_type(&self) -> Option<http::HeaderValue> {
         match self {
-            OtlpProtocol::HttpJson => libdd_common::header::APPLICATION_JSON,
-            OtlpProtocol::HttpProtobuf => libdd_common::header::APPLICATION_PROTOBUF,
+            OtlpProtocol::HttpJson => Some(libdd_common::header::APPLICATION_JSON),
+            OtlpProtocol::HttpProtobuf => Some(libdd_common::header::APPLICATION_PROTOBUF),
+            OtlpProtocol::Grpc => None,
         }
     }
 
-    /// Encode the prost OTLP request to this protocol's wire format. Crate-internal so the
-    /// third-party `serde_json::Error` does not leak into the public API.
     pub(crate) fn encode(
         &self,
         req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest,
-    ) -> Result<Vec<u8>, serde_json::Error> {
+    ) -> Option<Result<Vec<u8>, serde_json::Error>> {
         match self {
-            OtlpProtocol::HttpJson => libdd_trace_utils::otlp_encoder::encode_otlp_json(req),
-            OtlpProtocol::HttpProtobuf => {
-                Ok(libdd_trace_utils::otlp_encoder::encode_otlp_protobuf(req))
-            }
+            OtlpProtocol::HttpJson => Some(libdd_trace_utils::otlp_encoder::encode_otlp_json(req)),
+            OtlpProtocol::HttpProtobuf => Some(Ok(
+                libdd_trace_utils::otlp_encoder::encode_otlp_protobuf(req),
+            )),
+            OtlpProtocol::Grpc => None,
         }
     }
 }
@@ -86,8 +77,7 @@ pub struct OtlpTraceConfig {
 }
 
 /// Per-request OTLP gRPC trace exporter configuration.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
-#[allow(dead_code)]
+#[cfg(not(target_arch = "wasm32"))]
 #[derive(Clone, Debug)]
 pub struct OtlpGrpcTraceConfig {
     /// Custom key-value pairs forwarded as gRPC request metadata.
@@ -112,26 +102,21 @@ mod tests {
             OtlpProtocol::from_str("http/protobuf").unwrap(),
             OtlpProtocol::HttpProtobuf
         );
+        assert_eq!(OtlpProtocol::from_str("grpc").unwrap(), OtlpProtocol::Grpc);
         assert!(OtlpProtocol::from_str("nonsense").is_err());
     }
 
-    #[test]
-    fn grpc_is_rejected_at_parse() {
-        // gRPC is unsupported, so it must not parse into a protocol: an unsupported value can
-        // never be constructed.
-        assert!(OtlpProtocol::from_str("grpc").is_err());
-    }
-
     #[test]
     fn protocol_content_types() {
         assert_eq!(
             OtlpProtocol::HttpJson.content_type(),
-            libdd_common::header::APPLICATION_JSON
+            Some(libdd_common::header::APPLICATION_JSON)
         );
         assert_eq!(
             OtlpProtocol::HttpProtobuf.content_type(),
-            libdd_common::header::APPLICATION_PROTOBUF
+            Some(libdd_common::header::APPLICATION_PROTOBUF)
         );
+        assert_eq!(OtlpProtocol::Grpc.content_type(), None);
     }
 }
 
diff --git a/libdd-data-pipeline/src/otlp/exporter.rs b/libdd-data-pipeline/src/otlp/exporter.rs
index a05a04cc6a..14eb1c7399 100644
--- a/libdd-data-pipeline/src/otlp/exporter.rs
+++ b/libdd-data-pipeline/src/otlp/exporter.rs
@@ -18,7 +18,7 @@ use std::time::Duration;
 pub(crate) const OTLP_MAX_RETRIES: u32 = 4;
 /// No retries on shutdown to avoid a long backoff in the shutdown window.
 pub(crate) const OTLP_SHUTDOWN_MAX_RETRIES: u32 = 0;
-const OTLP_RETRY_DELAY_MS: u64 = 100;
+pub(crate) const OTLP_RETRY_DELAY_MS: u64 = 100;
 
 /// POST an OTLP HTTP payload to `endpoint_url` with the given `content_type` (callers pass JSON or
 /// protobuf); `test_token` enables snapshot tests.
@@ -125,13 +125,18 @@ pub async fn send_otlp_traces_http<C: HttpClientCapability + SleepCapability>(
     test_token: Option<&str>,
     body: Vec<u8>,
 ) -> Result<(), TraceExporterError> {
+    let content_type = config.protocol.content_type().ok_or_else(|| {
+        TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(
+            "OTLP gRPC protocol cannot be sent over the HTTP export path".to_string(),
+        ))
+    })?;
     send_otlp_http(
         capabilities,
         &config.endpoint_url,
         &config.headers,
         config.timeout,
         test_token,
-        config.protocol.content_type(),
+        content_type,
         body,
         OTLP_MAX_RETRIES,
     )
diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs
index ea96524204..41f77e47f7 100644
--- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs
+++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs
@@ -10,17 +10,20 @@
 use crate::otlp::config::OtlpGrpcTraceConfig;
 use crate::trace_exporter::error::{BuilderErrorKind, RequestError, TraceExporterError};
 use bytes::Bytes;
-use http_body_util::{BodyExt, Collected};
+use http_body_util::{BodyExt, Collected, Limited};
 use hyper::client::conn::http2;
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::{
-    ExportTraceServiceRequest, ExportTraceServiceResponse,
+    ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse,
 };
+use libdd_trace_utils::send_with_retry::TRACE_EXPORTER_USER_AGENT;
+use prost::Message as _;
 use std::error::Error as StdError;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
+use std::time::Duration;
 use tokio::net::TcpStream;
 use tonic::body::Body as TonicBody;
 use tonic::client::{Grpc, GrpcService};
@@ -29,12 +32,52 @@ use tonic::{Code, Request, Status};
 use tracing::warn;
 
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
+const MAX_GRPC_RESPONSE_SIZE: usize = 4 * 1024 * 1024;
+const RETRY_INFO_TYPE_URL: &str = "type.googleapis.com/google.rpc.RetryInfo";
+
+#[derive(Debug)]
+pub(crate) enum GrpcExportError {
+    Retryable {
+        error: TraceExporterError,
+        retry_after: Option<Duration>,
+    },
+    NonRetryable(TraceExporterError),
+}
+
+#[derive(Clone, PartialEq, prost::Message)]
+struct RpcStatusDetails {
+    #[prost(message, repeated, tag = "3")]
+    details: Vec<RpcStatusDetail>,
+}
+
+#[derive(Clone, PartialEq, prost::Message)]
+struct RpcStatusDetail {
+    #[prost(string, tag = "1")]
+    type_url: String,
+    #[prost(bytes = "vec", tag = "2")]
+    value: Vec<u8>,
+}
+
+#[derive(Clone, Copy, PartialEq, prost::Message)]
+struct RetryInfo {
+    #[prost(message, optional, tag = "1")]
+    retry_delay: Option<ProtoDuration>,
+}
+
+#[derive(Clone, Copy, PartialEq, prost::Message)]
+struct ProtoDuration {
+    #[prost(int64, tag = "1")]
+    seconds: i64,
+    #[prost(int32, tag = "2")]
+    nanos: i32,
+}
 
 /// tonic 0.14 moved `ProstCodec` to the separate `tonic-prost` crate; we hand-roll a minimal
 /// codec here to avoid that extra dependency and keep tonic at `default-features = false`.
 pub(crate) mod prost_codec {
     use prost::Message as ProstMessage;
     use std::marker::PhantomData;
+    use std::sync::Arc;
     use tonic::codec::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder};
     use tonic::Status;
 
@@ -53,16 +96,16 @@ pub(crate) mod prost_codec {
         // Shared with the `Encoder` impl below. tonic's `EncodeBuf`/`DecodeBuf` constructors are
         // private to the crate, so tests exercise this generic-over-`BufMut`/`Buf` core directly
         // instead of going through the `Encoder`/`Decoder` traits (see `codec_tests`).
-        fn encode_into(item: T, dst: &mut impl bytes::BufMut) -> Result<(), Status> {
+        fn encode_into(item: &T, dst: &mut impl bytes::BufMut) -> Result<(), Status> {
             item.encode(dst)
                 .map_err(|e| Status::internal(format!("Failed to encode protobuf message: {e}")))
         }
     }
-    impl<T: ProstMessage> Encoder for ProstEncoder<T> {
-        type Item = T;
+    impl<T: ProstMessage + Send + Sync + 'static> Encoder for ProstEncoder<T> {
+        type Item = Arc<T>;
         type Error = Status;
-        fn encode(&mut self, item: T, dst: &mut EncodeBuf<'_>) -> Result<(), Status> {
-            Self::encode_into(item, dst)
+        fn encode(&mut self, item: Arc<T>, dst: &mut EncodeBuf<'_>) -> Result<(), Status> {
+            Self::encode_into(item.as_ref(), dst)
         }
     }
 
@@ -89,10 +132,10 @@ pub(crate) mod prost_codec {
 
     impl<Enc, Dec> Codec for ProstCodecImpl<Enc, Dec>
     where
-        Enc: ProstMessage + Default + Send + 'static,
+        Enc: ProstMessage + Default + Send + Sync + 'static,
         Dec: ProstMessage + Default + Send + 'static,
     {
-        type Encode = Enc;
+        type Encode = Arc<Enc>;
         type Decode = Dec;
         type Encoder = ProstEncoder<Enc>;
         type Decoder = ProstDecoder<Dec>;
@@ -112,7 +155,6 @@ pub(crate) mod prost_codec {
             ExportTraceServiceRequest, ExportTraceServiceResponse,
         };
         use libdd_trace_protobuf::opentelemetry::proto::trace::v1::ResourceSpans;
-
         // Round-trips through the `BufMut`/`Buf`-generic core (`encode_into`/`decode_from`) that
         // the `Encoder`/`Decoder` impls delegate to, over a plain `BytesMut`; see
         // `ProstEncoder::encode_into` for why the `Codec` traits can't be driven directly.
@@ -126,7 +168,7 @@ pub(crate) mod prost_codec {
                 }],
             };
             let mut buf = BytesMut::new();
-            ProstEncoder::encode_into(msg.clone(), &mut buf).unwrap();
+            ProstEncoder::encode_into(&msg, &mut buf).unwrap();
             assert!(!buf.is_empty());
 
             let out = ProstDecoder::decode_from(&mut buf).unwrap();
@@ -172,7 +214,9 @@ impl GrpcService<TonicBody> for H2Service {
             let request = async move {
                 let resp = sender.send_request(req).await?;
                 let (parts, incoming) = resp.into_parts();
-                let collected = incoming.collect().await?;
+                let collected = Limited::new(incoming, MAX_GRPC_RESPONSE_SIZE)
+                    .collect()
+                    .await?;
                 Ok::<_, BoxError>(http::Response::from_parts(parts, collected))
             };
             tokio::pin!(conn);
@@ -182,17 +226,10 @@ impl GrpcService<TonicBody> for H2Service {
                 biased;
                 response = &mut request => response?,
                 connection = &mut conn => {
-                    connection?;
-                    return Err(std::io::Error::new(
-                        std::io::ErrorKind::UnexpectedEof,
-                        "gRPC connection closed before the response completed",
-                    ).into());
+                    let _ = connection;
+                    request.await?
                 }
             };
-
-            // The completed request has dropped its sender. Polling the connection to completion
-            // lets hyper shut it down gracefully; the exporter's outer timeout bounds this wait.
-            conn.await?;
             Ok(response)
         })
     }
@@ -202,7 +239,8 @@ impl GrpcService<TonicBody> for H2Service {
 /// service. Holds no live connection and thus no background task (nothing to rebuild across fork).
 #[derive(Clone, Debug)]
 pub(crate) struct OtlpGrpcTransport {
-    pub(crate) config: OtlpGrpcTraceConfig,
+    pub(crate) timeout: Duration,
+    pub(crate) otel_trace_semantics_enabled: bool,
     origin: http::Uri,
     service: H2Service,
     /// Custom headers parsed to gRPC metadata once at build time.
@@ -210,12 +248,15 @@ pub(crate) struct OtlpGrpcTransport {
 }
 
 /// Validate a gRPC endpoint (plaintext `http://` only) and build the transport.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
-#[allow(dead_code)]
 pub(crate) fn build_grpc_transport(
     endpoint_url: &str,
     config: OtlpGrpcTraceConfig,
 ) -> Result<OtlpGrpcTransport, TraceExporterError> {
+    let OtlpGrpcTraceConfig {
+        headers,
+        timeout,
+        otel_trace_semantics_enabled,
+    } = config;
     let uri = endpoint_url.parse::<http::Uri>()?;
 
     let scheme = uri.scheme().ok_or_else(|| {
@@ -244,8 +285,28 @@ pub(crate) fn build_grpc_transport(
             "gRPC endpoint must include an authority".to_string(),
         ))
     })?;
+    if authority.host().is_empty() {
+        return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri(
+            "gRPC endpoint authority must include a host".to_string(),
+        )));
+    }
+    let authority_text = authority.as_str();
+    if authority_text.contains('@') {
+        return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri(
+            "gRPC endpoint authority must not include userinfo".to_string(),
+        )));
+    }
+    let dial_authority = match authority.port_u16() {
+        Some(_) => authority_text.to_string(),
+        None if authority_text == authority.host() => format!("{}:80", authority.host()),
+        None => {
+            return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri(
+                "gRPC endpoint authority contains an invalid port".to_string(),
+            )));
+        }
+    };
     let service = H2Service {
-        authority: Arc::from(authority.as_str()),
+        authority: Arc::from(dial_authority),
     };
 
     // Origin = scheme+authority + normalized path prefix; tonic appends the RPC method path.
@@ -261,9 +322,8 @@ pub(crate) fn build_grpc_transport(
 
     // Parse custom headers to gRPC metadata once here rather than on every send. Invalid entries
     // are skipped with a single warning (logging only the key: a value may carry a secret).
-    let metadata_headers = config
-        .headers
-        .iter()
+    let metadata_headers = headers
+        .into_iter()
         .filter_map(|(k, v)| {
             match (
                 k.parse::<AsciiMetadataKey>(),
@@ -279,7 +339,8 @@ pub(crate) fn build_grpc_transport(
         .collect();
 
     Ok(OtlpGrpcTransport {
-        config,
+        timeout,
+        otel_trace_semantics_enabled,
         origin,
         service,
         metadata_headers,
@@ -293,14 +354,12 @@ type ExportCodec =
     prost_codec::ProstCodecImpl<ExportTraceServiceRequest, ExportTraceServiceResponse>;
 
 /// Send one OTLP trace export request over gRPC. Bounds connect + RPC with a single timeout.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
-#[allow(dead_code)]
 pub(crate) async fn send_otlp_traces_grpc(
     transport: &OtlpGrpcTransport,
     test_token: Option<&str>,
     client_computed_stats: bool,
-    request: ExportTraceServiceRequest,
-) -> Result<(), TraceExporterError> {
+    request: Arc<ExportTraceServiceRequest>,
+) -> Result<(), GrpcExportError> {
     let mut req = Request::new(request);
     attach_metadata(
         &mut req,
@@ -312,29 +371,55 @@ pub(crate) async fn send_otlp_traces_grpc(
     let path = http::uri::PathAndQuery::from_static(GRPC_EXPORT_PATH);
     let codec = ExportCodec::default();
 
-    tokio::time::timeout(transport.config.timeout, async {
+    tokio::time::timeout(transport.timeout, async {
         let mut client = Grpc::with_origin(transport.service.clone(), transport.origin.clone());
-        client.ready().await.map_err(|e| {
-            TraceExporterError::Io(std::io::Error::other(format!("gRPC not ready: {e}")))
-        })?;
         client
+            .ready()
+            .await
+            .map_err(|e| GrpcExportError::Retryable {
+                error: TraceExporterError::Io(std::io::Error::other(format!(
+                    "gRPC not ready: {e}"
+                ))),
+                retry_after: None,
+            })?;
+        let response = client
             .unary(req, path, codec)
             .await
-            .map(|_resp| ())
-            .map_err(grpc_status_to_error)
+            .map_err(grpc_status_to_error)?;
+        if let Some(details) = partial_success_details(response.get_ref()) {
+            warn!(
+                rejected_spans = details.rejected_spans,
+                error_message = %details.error_message,
+                "OTLP gRPC export was only partially accepted"
+            );
+        }
+        Ok(())
     })
     .await
-    .map_err(|_| TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)))?
+    .map_err(|_| GrpcExportError::Retryable {
+        error: TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),
+        retry_after: None,
+    })?
 }
 
-// Insert the pre-validated custom headers, the optional test-session token, and (when enabled) the
-// client-computed-stats marker into the request metadata.
-fn attach_metadata(
-    req: &mut Request<ExportTraceServiceRequest>,
+fn attach_metadata<T>(
+    req: &mut Request<T>,
     headers: &[(AsciiMetadataKey, AsciiMetadataValue)],
     test_token: Option<&str>,
     client_computed_stats: bool,
 ) {
+    req.metadata_mut().insert(
+        AsciiMetadataKey::from_static("user-agent"),
+        AsciiMetadataValue::from_static(TRACE_EXPORTER_USER_AGENT),
+    );
+    for (key, value) in libdd_common::entity_id::get_entity_headers() {
+        if let (Ok(key), Ok(value)) = (
+            key.parse::<AsciiMetadataKey>(),
+            value.parse::<AsciiMetadataValue>(),
+        ) {
+            req.metadata_mut().insert(key, value);
+        }
+    }
     for (key, val) in headers {
         req.metadata_mut().insert(key.clone(), val.clone());
     }
@@ -357,39 +442,98 @@ fn attach_metadata(
     }
 }
 
-fn grpc_status_to_error(status: Status) -> TraceExporterError {
-    // Transport/IO failures from our bare `H2Service` fall through to `Code::Unknown` with the
-    // original `std::io::Error` somewhere in the source chain (directly for a connect failure, or
-    // wrapped in a `hyper::Error` for a post-connect handshake/read/write failure). Walk the chain
-    // and recover it so these map to `Io` rather than an application-level `Request` error.
-    if status.code() == Code::Unknown {
-        let mut cause: Option<&(dyn std::error::Error + 'static)> = status.source();
-        while let Some(err) = cause {
-            if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
-                return TraceExporterError::Io(std::io::Error::new(
+fn partial_success_details(
+    response: &ExportTraceServiceResponse,
+) -> Option<&ExportTracePartialSuccess> {
+    response
+        .partial_success
+        .as_ref()
+        .filter(|details| details.rejected_spans != 0 || !details.error_message.is_empty())
+}
+
+fn grpc_status_to_error(status: Status) -> GrpcExportError {
+    let retry_after = retry_info_delay(&status);
+
+    // Recover local transport failures before classifying source-less remote gRPC statuses.
+    let mut cause: Option<&(dyn std::error::Error + 'static)> = status.source();
+    while let Some(err) = cause {
+        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
+            return GrpcExportError::Retryable {
+                error: TraceExporterError::Io(std::io::Error::new(
                     io_err.kind(),
                     io_err.to_string(),
-                ));
-            }
-            cause = err.source();
+                )),
+                retry_after,
+            };
+        }
+        let hyper_retryable = err
+            .downcast_ref::<hyper::Error>()
+            .is_some_and(|error| error.is_canceled() || error.is_closed());
+        let h2_retryable = err.downcast_ref::<h2::Error>().is_some_and(|error| {
+            matches!(
+                error.reason(),
+                Some(h2::Reason::REFUSED_STREAM | h2::Reason::CANCEL)
+            )
+        });
+        if hyper_retryable || h2_retryable {
+            return GrpcExportError::Retryable {
+                error: TraceExporterError::Io(std::io::Error::new(
+                    std::io::ErrorKind::ConnectionReset,
+                    err.to_string(),
+                )),
+                retry_after,
+            };
         }
+        cause = err.source();
     }
-    match status.code() {
-        Code::Unavailable => TraceExporterError::Io(std::io::Error::new(
-            std::io::ErrorKind::ConnectionRefused,
-            status.message(),
-        )),
-        Code::DeadlineExceeded => TraceExporterError::Io(std::io::Error::new(
-            std::io::ErrorKind::TimedOut,
-            status.message(),
-        )),
-        _ => TraceExporterError::Request(RequestError::new(
-            http::StatusCode::INTERNAL_SERVER_ERROR,
-            status.message(),
-        )),
+    let code = status.code();
+    let (http_status, retryable) = match code {
+        Code::Cancelled => (http::StatusCode::REQUEST_TIMEOUT, true),
+        Code::InvalidArgument => (http::StatusCode::BAD_REQUEST, false),
+        Code::OutOfRange => (http::StatusCode::BAD_REQUEST, true),
+        Code::DeadlineExceeded => (http::StatusCode::GATEWAY_TIMEOUT, true),
+        Code::NotFound => (http::StatusCode::NOT_FOUND, false),
+        Code::AlreadyExists => (http::StatusCode::CONFLICT, false),
+        Code::Aborted => (http::StatusCode::CONFLICT, true),
+        Code::PermissionDenied => (http::StatusCode::FORBIDDEN, false),
+        Code::ResourceExhausted => (http::StatusCode::TOO_MANY_REQUESTS, retry_after.is_some()),
+        Code::FailedPrecondition => (http::StatusCode::PRECONDITION_FAILED, false),
+        Code::Unauthenticated => (http::StatusCode::UNAUTHORIZED, false),
+        Code::Unavailable => (http::StatusCode::SERVICE_UNAVAILABLE, true),
+        Code::Unimplemented => (http::StatusCode::NOT_IMPLEMENTED, false),
+        Code::DataLoss => (http::StatusCode::INTERNAL_SERVER_ERROR, true),
+        _ => (http::StatusCode::INTERNAL_SERVER_ERROR, false),
+    };
+    let error = TraceExporterError::Request(RequestError::new(
+        http_status,
+        &format!("gRPC {code:?}: {}", status.message()),
+    ));
+    if retryable {
+        GrpcExportError::Retryable { error, retry_after }
+    } else {
+        GrpcExportError::NonRetryable(error)
     }
 }
 
+fn retry_info_delay(status: &Status) -> Option<Duration> {
+    let rich_status = RpcStatusDetails::decode(status.details()).ok()?;
+    rich_status.details.into_iter().find_map(|detail| {
+        if detail.type_url != RETRY_INFO_TYPE_URL {
+            return None;
+        }
+        let retry_info = RetryInfo::decode(detail.value.as_slice()).ok()?;
+        let Some(delay) = retry_info.retry_delay else {
+            return Some(Duration::ZERO);
+        };
+        let seconds = u64::try_from(delay.seconds).ok()?;
+        let nanos = u32::try_from(delay.nanos).ok()?;
+        if nanos >= 1_000_000_000 {
+            return None;
+        }
+        Some(Duration::new(seconds, nanos))
+    })
+}
+
 #[cfg(test)]
 mod build_tests {
     use super::*;
@@ -421,6 +565,49 @@ mod build_tests {
         assert!(build_grpc_transport("http://localhost:4317", cfg()).is_ok());
     }
 
+    #[test]
+    fn supplies_http_default_port_for_dialing() {
+        let transport = build_grpc_transport("http://collector", cfg()).unwrap();
+
+        assert_eq!(transport.service.authority.as_ref(), "collector:80");
+        assert_eq!(
+            transport
+                .origin
+                .authority()
+                .map(|authority| authority.as_str()),
+            Some("collector")
+        );
+
+        let ipv6_transport = build_grpc_transport("http://[::1]", cfg()).unwrap();
+        assert_eq!(ipv6_transport.service.authority.as_ref(), "[::1]:80");
+    }
+
+    #[test]
+    fn preserves_explicit_port_for_dialing() {
+        let transport = build_grpc_transport("http://collector:4317", cfg()).unwrap();
+        assert_eq!(transport.service.authority.as_ref(), "collector:4317");
+
+        let ipv6_transport = build_grpc_transport("http://[::1]:4317", cfg()).unwrap();
+        assert_eq!(ipv6_transport.service.authority.as_ref(), "[::1]:4317");
+    }
+
+    #[test]
+    fn rejects_invalid_dial_authorities() {
+        for endpoint in [
+            "http://collector:not-a-port",
+            "http://collector:99999",
+            "http://collector:",
+            "http://:4317",
+            "http://user@collector",
+            "http://user@collector:4317",
+        ] {
+            assert!(
+                build_grpc_transport(endpoint, cfg()).is_err(),
+                "accepted invalid endpoint {endpoint}"
+            );
+        }
+    }
+
     #[test]
     fn normalizes_origin_path_prefix() {
         let t = build_grpc_transport("http://localhost:4317/otel/", cfg()).unwrap();
@@ -432,13 +619,11 @@ mod build_tests {
         let config = OtlpGrpcTraceConfig {
             headers: vec![
                 ("good-key".to_string(), "ok".to_string()),
-                // Invalid metadata key (contains a space): skipped, not fatal to the build.
                 ("bad key".to_string(), "v".to_string()),
             ],
             timeout: Duration::from_secs(5),
             otel_trace_semantics_enabled: false,
         };
-        // A malformed header must not fail the build; it is dropped and the valid one retained.
         let t = build_grpc_transport("http://localhost:4317", config).unwrap();
         assert_eq!(t.metadata_headers.len(), 1);
         assert_eq!(t.metadata_headers[0].0.as_str(), "good-key");
@@ -452,7 +637,6 @@ mod integration_tests {
     use bytes::Bytes;
     use h2::server;
     use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest;
-    use prost::Message as _;
     use std::time::Duration;
     use tokio::net::TcpListener;
 
@@ -467,26 +651,91 @@ mod integration_tests {
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn connection_refused_maps_to_io() {
-        // Port 1 has no listener → connect fails.
         let transport = build_grpc_transport("http://127.0.0.1:1", cfg()).unwrap();
         let err = send_otlp_traces_grpc(
             &transport,
             None,
             false,
-            ExportTraceServiceRequest::default(),
+            Arc::new(ExportTraceServiceRequest::default()),
         )
         .await
         .unwrap_err();
-        assert!(matches!(err, TraceExporterError::Io(_)), "got: {err:?}");
+        assert!(
+            matches!(
+                err,
+                GrpcExportError::Retryable {
+                    error: TraceExporterError::Io(_),
+                    retry_after: None
+                }
+            ),
+            "got: {err:?}"
+        );
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn response_body_is_limited() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut conn = server::handshake(socket).await.unwrap();
+            let (_req, mut respond) = conn.accept().await.unwrap().unwrap();
+            let sender = tokio::spawn(async move {
+                let response = http::Response::builder().status(200).body(()).unwrap();
+                let mut stream = respond.send_response(response, false).unwrap();
+                let data = Bytes::from(vec![0; MAX_GRPC_RESPONSE_SIZE + 1]);
+                let mut offset = 0;
+                while offset < data.len() {
+                    stream.reserve_capacity((data.len() - offset).min(16 * 1024));
+                    let Some(Ok(capacity)) =
+                        std::future::poll_fn(|cx| stream.poll_capacity(cx)).await
+                    else {
+                        break;
+                    };
+                    let end = (offset + capacity.min(data.len() - offset)).min(data.len());
+                    if stream
+                        .send_data(data.slice(offset..end), end == data.len())
+                        .is_err()
+                    {
+                        break;
+                    }
+                    offset = end;
+                }
+            });
+            while conn.accept().await.is_some() {}
+            sender.await.unwrap();
+        });
+
+        let mut service = H2Service {
+            authority: Arc::from(addr.to_string()),
+        };
+        let request = http::Request::builder()
+            .uri("http://localhost/")
+            .body(TonicBody::empty())
+            .unwrap();
+        let error = service.call(request).await.unwrap_err();
+        server.await.unwrap();
+
+        let mut source: Option<&(dyn StdError + 'static)> = Some(error.as_ref());
+        let mut length_limited = false;
+        while let Some(error) = source {
+            if error
+                .downcast_ref::<http_body_util::LengthLimitError>()
+                .is_some()
+            {
+                length_limited = true;
+                break;
+            }
+            source = error.source();
+        }
+        assert!(length_limited, "got: {error:?}");
     }
 
-    // Minimal in-process gRPC server: accepts one unary Export call, decodes the request, and
-    // replies OK with an empty ExportTraceServiceResponse plus a `grpc-status` trailer.
     async fn run_one_shot_grpc_server(listener: TcpListener) -> ExportTraceServiceRequest {
         let (socket, _) = listener.accept().await.unwrap();
         let mut conn = server::handshake(socket).await.unwrap();
         let (req, mut respond) = conn.accept().await.unwrap().unwrap();
-        // Drain the gRPC-framed request body: 1-byte compression flag + 4-byte length + message.
         let mut body = req.into_body();
         let mut buf = Vec::new();
         while let Some(chunk) = body.data().await {
@@ -496,7 +745,6 @@ mod integration_tests {
         }
         let decoded = ExportTraceServiceRequest::decode(&buf[5..]).unwrap();
 
-        // Respond: headers, one empty framed message, then grpc-status trailer.
         let resp = http::Response::builder()
             .status(200)
             .header("content-type", "application/grpc")
@@ -506,15 +754,12 @@ mod integration_tests {
         let msg = ExportTraceServiceResponse::default();
         let mut framed = vec![0u8; 5];
         msg.encode(&mut framed).unwrap();
-        let len = (framed.len() - 5) as u32;
+        let len = u32::try_from(framed.len() - 5).expect("response exceeds gRPC frame length");
         framed[1..5].copy_from_slice(&len.to_be_bytes());
         send.send_data(Bytes::from(framed), false).unwrap();
         let mut trailers = http::HeaderMap::new();
         trailers.insert("grpc-status", "0".parse().unwrap());
         send.send_trailers(trailers).unwrap();
-        // Keep the connection future alive (driving any remaining H2 protocol traffic, e.g. the
-        // client's final stream-closing frames) until the client drops its side and this
-        // `accept()` resolves to `None`, so the connection task doesn't outlive the test.
         let _ = tokio::time::timeout(Duration::from_secs(2), conn.accept()).await;
         decoded
     }
@@ -530,7 +775,7 @@ mod integration_tests {
         let mut request = ExportTraceServiceRequest::default();
         request.resource_spans.push(Default::default());
 
-        send_otlp_traces_grpc(&transport, None, false, request.clone())
+        send_otlp_traces_grpc(&transport, None, false, Arc::new(request.clone()))
             .await
             .expect("send should succeed");
 
@@ -538,13 +783,95 @@ mod integration_tests {
         assert_eq!(decoded.resource_spans.len(), 1);
     }
 
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn successful_response_wins_over_connection_teardown() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut conn = server::handshake(socket).await.unwrap();
+            let (req, mut respond) = conn.accept().await.unwrap().unwrap();
+            let mut body = req.into_body();
+            while let Some(chunk) = body.data().await {
+                let chunk = chunk.unwrap();
+                body.flow_control().release_capacity(chunk.len()).unwrap();
+            }
+
+            let response = http::Response::builder()
+                .status(200)
+                .header("content-type", "application/grpc")
+                .body(())
+                .unwrap();
+            let mut send = respond.send_response(response, false).unwrap();
+            send.send_data(Bytes::from_static(&[0, 0, 0, 0, 0]), false)
+                .unwrap();
+            let mut trailers = http::HeaderMap::new();
+            trailers.insert("grpc-status", "0".parse().unwrap());
+            send.send_trailers(trailers).unwrap();
+            let flush_deadline = tokio::time::sleep(Duration::from_millis(25));
+            tokio::pin!(flush_deadline);
+            tokio::select! {
+                _ = &mut flush_deadline => {}
+                _ = conn.accept() => {}
+            }
+            conn.abrupt_shutdown(h2::Reason::INTERNAL_ERROR);
+            while conn.accept().await.is_some() {}
+        });
+
+        let transport = build_grpc_transport(&format!("http://{addr}"), cfg()).unwrap();
+        send_otlp_traces_grpc(
+            &transport,
+            None,
+            false,
+            Arc::new(ExportTraceServiceRequest::default()),
+        )
+        .await
+        .expect("completed response should remain successful");
+        server.await.unwrap();
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn refused_stream_is_retryable() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut conn = server::handshake(socket).await.unwrap();
+            let (_req, mut respond) = conn.accept().await.unwrap().unwrap();
+            respond.send_reset(h2::Reason::REFUSED_STREAM);
+            while conn.accept().await.is_some() {}
+        });
+
+        let transport = build_grpc_transport(&format!("http://{addr}"), cfg()).unwrap();
+        let error = send_otlp_traces_grpc(
+            &transport,
+            None,
+            false,
+            Arc::new(ExportTraceServiceRequest::default()),
+        )
+        .await
+        .unwrap_err();
+        server.await.unwrap();
+
+        assert!(
+            matches!(
+                error,
+                GrpcExportError::Retryable {
+                    error: TraceExporterError::Io(_),
+                    retry_after: None
+                }
+            ),
+            "expected retryable transport error, got: {error:?}"
+        );
+    }
+
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn timeout_maps_to_io_timedout() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let addr = listener.local_addr().unwrap();
-        // Accept the connection but never drive the H2 handshake or send a response, so the send
-        // blocks until the timeout fires rather than completing or failing fast.
         let server = tokio::spawn(async move {
             let (socket, _) = listener.accept().await.unwrap();
             tokio::time::sleep(Duration::from_secs(30)).await;
@@ -561,16 +888,19 @@ mod integration_tests {
             &transport,
             None,
             false,
-            ExportTraceServiceRequest::default(),
+            Arc::new(ExportTraceServiceRequest::default()),
         )
         .await
         .unwrap_err();
         server.abort();
         match err {
-            TraceExporterError::Io(e) => {
+            GrpcExportError::Retryable {
+                error: TraceExporterError::Io(e),
+                retry_after: None,
+            } => {
                 assert_eq!(e.kind(), std::io::ErrorKind::TimedOut, "got: {e:?}")
             }
-            other => panic!("expected Io(TimedOut), got {other:?}"),
+            other => panic!("expected retryable Io(TimedOut), got {other:?}"),
         }
     }
 
@@ -579,28 +909,17 @@ mod integration_tests {
     async fn post_connect_transport_failure_maps_to_io() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let addr = listener.local_addr().unwrap();
-        // Complete the H2 handshake and accept the request, then abort the TCP connection with a
-        // RST (SO_LINGER=0) instead of a graceful close or a gRPC-status trailer. The client's
-        // in-flight read then fails with ECONNRESET. This is the FIX-1 case: the resulting
-        // `std::io::Error` is wrapped inside a `hyper::Error` below the tonic `Status`, so it is
-        // recovered only by walking the source chain in `grpc_status_to_error` — a graceful FIN
-        // instead surfaces as an h2 "canceled" status that would be misreported as `Request`.
         let server = tokio::spawn(async move {
             let (socket, _) = listener.accept().await.unwrap();
-            // Force a RST on close rather than a graceful FIN. `set_linger` is deprecated because a
-            // non-zero linger blocks the thread on drop; a zero duration instead drops buffered
-            // data and sends the RST immediately without blocking, which is exactly what we want.
             #[allow(deprecated)]
             socket.set_linger(Some(Duration::ZERO)).unwrap();
             let mut conn = server::handshake(socket).await.unwrap();
             let (req, _respond) = conn.accept().await.unwrap().unwrap();
-            // Drain the request body so the client is parked awaiting the response when we reset.
             let mut body = req.into_body();
             while let Some(chunk) = body.data().await {
                 let chunk = chunk.unwrap();
                 body.flow_control().release_capacity(chunk.len()).unwrap();
             }
-            // Drop the connection (and its socket) -> RST.
             drop(conn);
         });
 
@@ -609,13 +928,19 @@ mod integration_tests {
             &transport,
             None,
             false,
-            ExportTraceServiceRequest::default(),
+            Arc::new(ExportTraceServiceRequest::default()),
         )
         .await
         .unwrap_err();
         server.await.unwrap();
         assert!(
-            matches!(err, TraceExporterError::Io(_)),
+            matches!(
+                err,
+                GrpcExportError::Retryable {
+                    error: TraceExporterError::Io(_),
+                    retry_after: None
+                }
+            ),
             "expected Io (post-connect transport failure), got: {err:?}"
         );
     }
@@ -625,38 +950,143 @@ mod integration_tests {
 mod send_tests {
     use super::*;
     use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest;
+    use std::time::Duration;
     use tonic::{Code, Request, Status};
 
+    #[derive(Clone, PartialEq, prost::Message)]
+    struct TestRpcStatus {
+        #[prost(message, repeated, tag = "3")]
+        details: Vec<TestAny>,
+    }
+
+    #[derive(Clone, PartialEq, prost::Message)]
+    struct TestAny {
+        #[prost(string, tag = "1")]
+        type_url: String,
+        #[prost(bytes = "vec", tag = "2")]
+        value: Vec<u8>,
+    }
+
+    #[derive(Clone, Copy, PartialEq, prost::Message)]
+    struct TestRetryInfo {
+        #[prost(message, optional, tag = "1")]
+        retry_delay: Option<TestDuration>,
+    }
+
+    #[derive(Clone, Copy, PartialEq, prost::Message)]
+    struct TestDuration {
+        #[prost(int64, tag = "1")]
+        seconds: i64,
+        #[prost(int32, tag = "2")]
+        nanos: i32,
+    }
+
+    fn status_with_retry_info(code: Code, retry_delay: Duration) -> Status {
+        let retry_info = TestRetryInfo {
+            retry_delay: Some(TestDuration {
+                seconds: i64::try_from(retry_delay.as_secs()).unwrap(),
+                nanos: i32::try_from(retry_delay.subsec_nanos()).unwrap(),
+            }),
+        };
+        let rich_status = TestRpcStatus {
+            details: vec![TestAny {
+                type_url: RETRY_INFO_TYPE_URL.to_string(),
+                value: retry_info.encode_to_vec(),
+            }],
+        };
+        Status::with_details(
+            code,
+            "retry later",
+            Bytes::from(rich_status.encode_to_vec()),
+        )
+    }
+
     #[test]
-    fn status_transient_maps_to_io_kind() {
-        for (s, want) in [
-            (
-                Status::unavailable("down"),
-                std::io::ErrorKind::ConnectionRefused,
-            ),
-            (
-                Status::deadline_exceeded("slow"),
-                std::io::ErrorKind::TimedOut,
-            ),
+    fn resource_exhausted_with_retry_info_is_retryable() {
+        let retry_after = Duration::new(3, 250_000_000);
+
+        match grpc_status_to_error(status_with_retry_info(Code::ResourceExhausted, retry_after)) {
+            GrpcExportError::Retryable {
+                error: TraceExporterError::Request(error),
+                retry_after: actual_retry_after,
+            } => {
+                assert_eq!(error.status(), http::StatusCode::TOO_MANY_REQUESTS);
+                assert_eq!(actual_retry_after, Some(retry_after));
+            }
+            other => panic!("expected retryable throttling error, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn unavailable_honors_retry_info_delay() {
+        let retry_after = Duration::from_secs(7);
+
+        match grpc_status_to_error(status_with_retry_info(Code::Unavailable, retry_after)) {
+            GrpcExportError::Retryable {
+                error: TraceExporterError::Request(error),
+                retry_after: actual_retry_after,
+            } => {
+                assert_eq!(error.status(), http::StatusCode::SERVICE_UNAVAILABLE);
+                assert_eq!(error.msg(), "gRPC Unavailable: retry later");
+                assert_eq!(actual_retry_after, Some(retry_after));
+            }
+            other => panic!("expected throttled unavailable error, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn status_transient_remains_retryable_request_error() {
+        for (code, http_status) in [
+            (Code::Unavailable, http::StatusCode::SERVICE_UNAVAILABLE),
+            (Code::DeadlineExceeded, http::StatusCode::GATEWAY_TIMEOUT),
+            (Code::Cancelled, http::StatusCode::REQUEST_TIMEOUT),
+            (Code::Aborted, http::StatusCode::CONFLICT),
+            (Code::OutOfRange, http::StatusCode::BAD_REQUEST),
+            (Code::DataLoss, http::StatusCode::INTERNAL_SERVER_ERROR),
         ] {
-            match grpc_status_to_error(s) {
-                TraceExporterError::Io(e) => assert_eq!(e.kind(), want),
-                other => panic!("expected Io, got {other:?}"),
+            match grpc_status_to_error(Status::new(code, "transient")) {
+                GrpcExportError::Retryable {
+                    error: TraceExporterError::Request(error),
+                    retry_after: None,
+                } => {
+                    assert_eq!(error.status(), http_status);
+                    assert_eq!(error.msg(), format!("gRPC {code:?}: transient"));
+                }
+                other => panic!("expected retryable Request, got {other:?}"),
             }
         }
     }
 
     #[test]
     fn status_unknown_with_io_source_maps_to_io() {
-        // Exercises the `Code::Unknown` + io-source recovery path in `grpc_status_to_error`.
         let io_err = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused");
         let status = Status::from_error(Box::new(io_err));
         assert_eq!(status.code(), Code::Unknown);
         match grpc_status_to_error(status) {
-            TraceExporterError::Io(e) => {
+            GrpcExportError::Retryable {
+                error: TraceExporterError::Io(e),
+                retry_after: None,
+            } => {
                 assert_eq!(e.kind(), std::io::ErrorKind::ConnectionRefused)
             }
-            other => panic!("expected Io, got {other:?}"),
+            other => panic!("expected retryable Io, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn status_unavailable_with_io_source_maps_to_io() {
+        let io_err = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset");
+        let mut status = Status::unavailable("transport failure");
+        status.set_source(Arc::new(io_err));
+        assert_eq!(status.code(), Code::Unavailable);
+        match grpc_status_to_error(status) {
+            GrpcExportError::Retryable {
+                error: TraceExporterError::Io(e),
+                retry_after: None,
+            } => {
+                assert_eq!(e.kind(), std::io::ErrorKind::ConnectionReset)
+            }
+            other => panic!("expected retryable Io, got {other:?}"),
         }
     }
 
@@ -664,16 +1094,62 @@ mod send_tests {
     fn status_unknown_without_io_source_maps_to_request() {
         assert!(matches!(
             grpc_status_to_error(Status::new(Code::Unknown, "mystery")),
-            TraceExporterError::Request(_)
+            GrpcExportError::NonRetryable(TraceExporterError::Request(_))
         ));
     }
 
     #[test]
     fn status_application_errors_map_to_request() {
-        assert!(matches!(
-            grpc_status_to_error(Status::new(Code::Internal, "boom")),
-            TraceExporterError::Request(_)
-        ));
+        for (code, http_status) in [
+            (Code::InvalidArgument, http::StatusCode::BAD_REQUEST),
+            (Code::Unauthenticated, http::StatusCode::UNAUTHORIZED),
+            (Code::PermissionDenied, http::StatusCode::FORBIDDEN),
+            (Code::NotFound, http::StatusCode::NOT_FOUND),
+            (Code::AlreadyExists, http::StatusCode::CONFLICT),
+            (Code::ResourceExhausted, http::StatusCode::TOO_MANY_REQUESTS),
+            (
+                Code::FailedPrecondition,
+                http::StatusCode::PRECONDITION_FAILED,
+            ),
+            (Code::Unimplemented, http::StatusCode::NOT_IMPLEMENTED),
+            (Code::Internal, http::StatusCode::INTERNAL_SERVER_ERROR),
+        ] {
+            match grpc_status_to_error(Status::new(code, "failure")) {
+                GrpcExportError::NonRetryable(TraceExporterError::Request(error)) => {
+                    assert_eq!(error.status(), http_status);
+                    assert_eq!(error.msg(), format!("gRPC {code:?}: failure"));
+                }
+                other => panic!("expected Request, got {other:?}"),
+            }
+        }
+    }
+
+    #[test]
+    fn partial_success_details_ignores_empty_response() {
+        let empty = ExportTraceServiceResponse::default();
+        assert!(partial_success_details(&empty).is_none());
+
+        let present_but_empty = ExportTraceServiceResponse {
+            partial_success: Some(
+                libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTracePartialSuccess::default(),
+            ),
+        };
+        assert!(partial_success_details(&present_but_empty).is_none());
+    }
+
+    #[test]
+    fn partial_success_details_returns_rejections_and_warnings() {
+        let response = ExportTraceServiceResponse {
+            partial_success: Some(
+                libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTracePartialSuccess {
+                    rejected_spans: 3,
+                    error_message: "too many spans".to_string(),
+                },
+            ),
+        };
+        let details = partial_success_details(&response).unwrap();
+        assert_eq!(details.rejected_spans, 3);
+        assert_eq!(details.error_message, "too many spans");
     }
 
     #[test]
diff --git a/libdd-data-pipeline/src/otlp/mod.rs b/libdd-data-pipeline/src/otlp/mod.rs
index 7911616e90..ab82f314af 100644
--- a/libdd-data-pipeline/src/otlp/mod.rs
+++ b/libdd-data-pipeline/src/otlp/mod.rs
@@ -1,37 +1,10 @@
 // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-//! OTLP trace and trace-metrics export for libdatadog.
-//!
-//! When an OTLP endpoint is configured via
-//! [`crate::trace_exporter::TraceExporterBuilder::set_otlp_endpoint`], the trace exporter sends
-//! traces in OTLP HTTP format to that endpoint instead of the Datadog agent; the wire encoding
-//! (JSON or protobuf) is selected via [`OtlpProtocol`]. The host language is responsible for
-//! resolving the endpoint from its own configuration (e.g.
-//! `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`).
-//!
-//! With [`set_otlp_metrics_endpoint`](crate::trace_exporter::TraceExporterBuilder::set_otlp_metrics_endpoint),
-//! client-computed span stats ship as a `traces.span.sdk.metrics.duration` OTLP histogram
-//! instead of going to the agent `/v0.6/stats` endpoint.
-//!
-//! ## Sampling
-//!
-//! The exporter enforces the sampling decision already made by the tracer: unsampled chunks are
-//! dropped via `drop_chunks` before export. It does not apply its own sampling policy. The tracer
-//! (e.g. dd-trace-py) is responsible for inheriting the sampling decision from the distributed
-//! trace context; when no decision is present, the tracer typically uses 100% (always on).
-//!
-//! ## Partial flush
-//!
-//! For the POC, partial flush is disabled. The tracer should only invoke the exporter when all
-//! spans from a local trace are closed (i.e. send complete trace chunks). This crate does not
-//! buffer or flush partially—it exports whatever trace chunks it receives.
-
 pub mod config;
 pub mod exporter;
 pub mod metrics;
 
-// gRPC OTLP export depends on tonic/hyper, which do not build for wasm32.
 #[cfg(not(target_arch = "wasm32"))]
 pub mod grpc_exporter;
 
@@ -40,3 +13,10 @@ pub use config::{OtlpMetricsConfig, OtlpProtocol, OtlpTraceConfig};
 pub use exporter::send_otlp_traces_http;
 pub use libdd_trace_utils::otlp_encoder::{map_traces_to_otlp, OtlpResourceInfo};
 pub use metrics::OtlpStatsExporter;
+
+#[cfg(not(target_arch = "wasm32"))]
+pub use config::OtlpGrpcTraceConfig;
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) use grpc_exporter::{
+    build_grpc_transport, send_otlp_traces_grpc, GrpcExportError, OtlpGrpcTransport,
+};
diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs
index ad79a8b756..4f7ad40fea 100644
--- a/libdd-data-pipeline/src/trace_exporter/builder.rs
+++ b/libdd-data-pipeline/src/trace_exporter/builder.rs
@@ -4,6 +4,8 @@
 use crate::agent_info::AgentInfoFetcher;
 use crate::agentless::config::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT};
 use crate::otlp::config::{OtlpProtocol, DEFAULT_OTLP_TIMEOUT};
+#[cfg(not(target_arch = "wasm32"))]
+use crate::otlp::{build_grpc_transport, OtlpGrpcTraceConfig};
 use crate::otlp::{OtlpMetricsConfig, OtlpResourceInfo, OtlpTraceConfig};
 #[cfg(feature = "telemetry")]
 use crate::telemetry::TelemetryClientBuilder;
@@ -16,8 +18,9 @@ use crate::trace_exporter::TelemetryConfig;
 use crate::trace_exporter::TelemetryInstrumentationSessions;
 use crate::trace_exporter::TraceExporterWorkers;
 use crate::trace_exporter::{
-    add_path, StatsComputationStatus, TraceExporter, TraceExporterError, TraceExporterInputFormat,
-    TraceExporterOutputFormat, TraceSerializer, TracerMetadata, INFO_ENDPOINT,
+    add_path, OtlpExportMode, StatsComputationStatus, TraceExporter, TraceExporterError,
+    TraceExporterInputFormat, TraceExporterOutputFormat, TraceSerializer, TracerMetadata,
+    INFO_ENDPOINT,
 };
 use arc_swap::ArcSwap;
 #[cfg(feature = "telemetry")]
@@ -453,11 +456,12 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         self
     }
 
-    /// Enables OTLP HTTP/JSON export and sets the endpoint URL.
+    /// Enables OTLP trace export and sets the endpoint URL.
     ///
-    /// When set, traces are sent to this endpoint in OTLP HTTP/JSON format instead of the
-    /// Datadog agent. The host language is responsible for resolving the endpoint from its
-    /// configuration (e.g. `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this method.
+    /// When set, traces are sent to this endpoint using the protocol selected by
+    /// [`Self::set_otlp_protocol`] instead of the Datadog agent. The host language is responsible
+    /// for resolving the endpoint from its configuration (e.g.
+    /// `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this method.
     ///
     /// OTLP trace export is mutually exclusive with agentless trace export
     /// ([`Self::set_agentless_endpoint`]); configuring both causes
@@ -472,16 +476,19 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
         self
     }
 
-    /// Selects the OTLP export protocol: [`OtlpProtocol::HttpJson`] (default) or
-    /// [`OtlpProtocol::HttpProtobuf`]. The host language resolves this from
-    /// `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` / `OTEL_EXPORTER_OTLP_PROTOCOL`; a `grpc` value is
-    /// unsupported and is rejected when parsed into [`OtlpProtocol`], so it never reaches here.
+    /// Selects the OTLP export protocol: [`OtlpProtocol::HttpJson`] (default),
+    /// [`OtlpProtocol::HttpProtobuf`], or [`OtlpProtocol::Grpc`]. The host language resolves this
+    /// from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` / `OTEL_EXPORTER_OTLP_PROTOCOL`; all three OTel
+    /// protocol strings (`http/json`, `http/protobuf`, `grpc`) parse into [`OtlpProtocol`]. gRPC
+    /// export requires a plaintext `http://` endpoint: an `https://` gRPC endpoint is rejected at
+    /// [`build`](Self::build) time, and gRPC is not supported on wasm32 targets (also rejected at
+    /// build time).
     pub fn set_otlp_protocol(&mut self, protocol: OtlpProtocol) -> &mut Self {
         self.otlp_protocol = protocol;
         self
     }
 
-    /// Sets additional HTTP headers to include in OTLP trace export requests.
+    /// Sets additional headers or gRPC metadata to include in OTLP trace export requests.
     ///
     /// Headers should be provided as key-value pairs. The host language is responsible for
     /// resolving headers from its configuration (e.g. `OTEL_EXPORTER_OTLP_TRACES_HEADERS`)
@@ -656,6 +663,31 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
 
         self.validate_export_targets()?;
 
+        let otlp_timeout = self
+            .connection_timeout
+            .map(Duration::from_millis)
+            .unwrap_or(DEFAULT_OTLP_TIMEOUT);
+        #[cfg(not(target_arch = "wasm32"))]
+        let grpc_transport = match self.otlp_endpoint.as_deref() {
+            Some(url) if self.otlp_protocol == OtlpProtocol::Grpc => Some(build_grpc_transport(
+                url,
+                OtlpGrpcTraceConfig {
+                    headers: self.otlp_headers.clone(),
+                    timeout: otlp_timeout,
+                    otel_trace_semantics_enabled: self.otel_trace_semantics_enabled,
+                },
+            )?),
+            _ => None,
+        };
+        #[cfg(target_arch = "wasm32")]
+        if self.otlp_endpoint.is_some() && self.otlp_protocol == OtlpProtocol::Grpc {
+            return Err(TraceExporterError::Builder(
+                BuilderErrorKind::InvalidConfiguration(
+                    "OTLP gRPC export is not supported on wasm32 targets".to_string(),
+                ),
+            ));
+        }
+
         let shared_runtime = match self.shared_runtime {
             Some(rt) => rt,
             None => Arc::new(R::new().map_err(|e| {
@@ -795,22 +827,31 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             _ => None,
         };
 
-        let otlp_timeout = self
-            .connection_timeout
-            .map(Duration::from_millis)
-            .unwrap_or(DEFAULT_OTLP_TIMEOUT);
+        let otlp_headers = self.otlp_headers;
+        let instrumentation_scope_name = self.instrumentation_scope_name;
+        let instrumentation_scope_version = self.instrumentation_scope_version;
 
-        // `self.otlp_protocol` is always an HTTP encoding here: gRPC is rejected at the parse
-        // boundary (`OtlpProtocol::from_str`) and so can never be constructed.
-        let otlp_config = otlp_endpoint.map(|url| OtlpTraceConfig {
-            endpoint_url: url,
-            headers: build_otlp_header_map(self.otlp_headers),
-            timeout: otlp_timeout,
-            protocol: self.otlp_protocol,
-            instrumentation_scope_name: self.instrumentation_scope_name,
-            instrumentation_scope_version: self.instrumentation_scope_version,
-            otel_trace_semantics_enabled: self.otel_trace_semantics_enabled,
-        });
+        let otlp = match otlp_endpoint {
+            #[cfg(not(target_arch = "wasm32"))]
+            Some(_) if self.otlp_protocol == OtlpProtocol::Grpc => {
+                let transport = grpc_transport.ok_or_else(|| {
+                    TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration(
+                        "OTLP gRPC transport was not initialized".to_string(),
+                    ))
+                })?;
+                Some(OtlpExportMode::Grpc(transport))
+            }
+            Some(url) => Some(OtlpExportMode::Http(OtlpTraceConfig {
+                endpoint_url: url,
+                headers: build_otlp_header_map(otlp_headers),
+                timeout: otlp_timeout,
+                protocol: self.otlp_protocol,
+                instrumentation_scope_name: instrumentation_scope_name.clone(),
+                instrumentation_scope_version: instrumentation_scope_version.clone(),
+                otel_trace_semantics_enabled: self.otel_trace_semantics_enabled,
+            })),
+            None => None,
+        };
 
         let otlp_metrics_config = self.otlp_metrics_endpoint.map(|url| OtlpMetricsConfig {
             endpoint_url: url,
@@ -833,6 +874,17 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             .runtime_id
             .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
 
+        let base_otlp_resource = |rid: &str| {
+            let mut r = OtlpResourceInfo::default();
+            r.service = self.service.clone();
+            r.env = self.env.clone();
+            r.app_version = self.app_version.clone();
+            r.language = self.language.clone();
+            r.tracer_version = self.tracer_version.clone();
+            r.runtime_id = rid.to_string();
+            r
+        };
+
         // OTLP metrics + stats bucket size: start the concentrator unconditionally (bypass the
         // agent gate) so `check_agent_info` cannot later disable stats.
         let mut otlp_stats_enabled = false;
@@ -856,13 +908,7 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
                 #[cfg(feature = "stats-obfuscation")]
                 Some(stats_obfuscation_config.clone()),
             )));
-            let mut resource = OtlpResourceInfo::default();
-            resource.service = self.service.clone();
-            resource.env = self.env.clone();
-            resource.app_version = self.app_version.clone();
-            resource.language = self.language.clone();
-            resource.tracer_version = self.tracer_version.clone();
-            resource.runtime_id = runtime_id.clone();
+            let mut resource = base_otlp_resource(&runtime_id);
             resource.hostname = self.hostname.clone();
             resource.process_tags = self.process_tags.clone();
             resource.tracer_tags = self.tracer_tags.clone();
@@ -1011,6 +1057,25 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             };
         }
 
+        let otlp_resource_info = if let Some(mode) = otlp.as_ref() {
+            let mut r = base_otlp_resource(&runtime_id);
+            r.client_computed_stats = self.client_computed_stats || otlp_stats_enabled;
+            match mode {
+                OtlpExportMode::Http(config) => {
+                    r.instrumentation_scope_name = config.instrumentation_scope_name.clone();
+                    r.instrumentation_scope_version = config.instrumentation_scope_version.clone();
+                }
+                #[cfg(not(target_arch = "wasm32"))]
+                OtlpExportMode::Grpc(_) => {
+                    r.instrumentation_scope_name = instrumentation_scope_name;
+                    r.instrumentation_scope_version = instrumentation_scope_version;
+                }
+            }
+            r
+        } else {
+            OtlpResourceInfo::default()
+        };
+
         let log_output = self
             .output_to_log
             .then(|| self.log_max_line_size.unwrap_or(DEFAULT_LOG_MAX_LINE_SIZE));
@@ -1074,7 +1139,8 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
             agent_payload_response_version: self
                 .agent_rates_payload_version_enabled
                 .then(AgentResponsePayloadVersion::new),
-            otlp_config,
+            otlp,
+            otlp_resource_info,
             agentless_config,
             trace_filterer: ArcSwap::from_pointee(TraceFilterer::with_empty_conf()),
             otlp_stats_enabled,
@@ -1084,23 +1150,6 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
     }
 
     /// Reject configurations that combine mutually exclusive trace export targets.
-    ///
-    /// Trace export uses exactly one of three transports:
-    /// - the Datadog Agent (via [`Self::set_url`], the default when no transport is set),
-    /// - an OTLP HTTP/JSON endpoint (via [`Self::set_otlp_endpoint`]), or
-    /// - the agentless intake (via [`Self::set_agentless_endpoint`]).
-    ///
-    /// Exclusion rules enforced here:
-    /// - OTLP and agentless cannot both be configured.
-    /// - Agentless cannot be combined with a caller-supplied agent URL.
-    /// - Log output cannot be combined with OTLP or agentless trace export.
-    /// - [`Self::set_agentless_timeout`] requires [`Self::set_agentless_endpoint`].
-    /// - [`Self::set_agentless_stats_endpoint`] requires agentless trace export
-    ///   ([`Self::set_agentless_endpoint`]) and is incompatible with OTLP stats
-    ///   ([`Self::set_otlp_metrics_endpoint`]).
-    ///
-    /// OTLP and an agent URL may coexist: the agent URL is still useful for auxiliary
-    /// agent endpoints (info, stats) even when trace payloads are routed to OTLP.
     fn validate_export_targets(&self) -> Result<(), TraceExporterError> {
         let otlp_set = self.otlp_endpoint.is_some();
         let agentless_set = self.agentless_endpoint.is_some();
@@ -1237,7 +1286,9 @@ impl<R: SharedRuntime> TraceExporterBuilder<R> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::agent_info;
     use crate::trace_exporter::error::BuilderErrorKind;
+    use httpmock::prelude::*;
     use libdd_capabilities_impl::NativeCapabilities;
     use libdd_shared_runtime::ForkSafeRuntime;
 
@@ -1314,7 +1365,13 @@ mod tests {
         assert_eq!(exporter.metadata.language_interpreter_vendor, "node");
         assert_eq!(exporter.metadata.git_commit_sha, "797e9ea");
         assert!(exporter.metadata.client_computed_stats);
-        let otlp_config = exporter.otlp_config.as_ref().unwrap();
+        #[cfg(not(target_arch = "wasm32"))]
+        let otlp_config = match exporter.otlp.as_ref().unwrap() {
+            OtlpExportMode::Http(c) => c,
+            OtlpExportMode::Grpc(_) => panic!("expected HTTP OTLP mode"),
+        };
+        #[cfg(target_arch = "wasm32")]
+        let OtlpExportMode::Http(otlp_config) = exporter.otlp.as_ref().unwrap();
         assert_eq!(otlp_config.instrumentation_scope_name, "dd-trace-js");
         assert_eq!(otlp_config.instrumentation_scope_version, "7.0.0-pre");
         assert!(!exporter.restart_after_fork);
@@ -1590,6 +1647,72 @@ mod tests {
         );
     }
 
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn build_with_grpc_protocol_and_endpoint_succeeds() {
+        let mut builder = TraceExporterBuilder::default();
+        builder
+            .set_otlp_endpoint("http://localhost:4317")
+            .set_otlp_protocol(OtlpProtocol::Grpc)
+            .set_otlp_instrumentation_scope("dd-trace-js", "7.0.0-pre");
+        let exporter = builder.build::<NativeCapabilities>().unwrap();
+        assert!(matches!(exporter.otlp, Some(OtlpExportMode::Grpc(_))));
+        assert_eq!(
+            exporter.otlp_resource_info.instrumentation_scope_name,
+            "dd-trace-js"
+        );
+        assert_eq!(
+            exporter.otlp_resource_info.instrumentation_scope_version,
+            "7.0.0-pre"
+        );
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn invalid_grpc_endpoint_does_not_start_info_worker() {
+        agent_info::clear_cache_for_test();
+        let server = MockServer::start();
+        let info = server.mock(|when, then| {
+            when.method(GET).path("/info");
+            then.status(200).body("{}");
+        });
+        let shared_runtime = Arc::new(ForkSafeRuntime::new().unwrap());
+        let mut builder = TraceExporterBuilder::default();
+        builder
+            .set_shared_runtime(shared_runtime.clone())
+            .set_url(&server.base_url())
+            .set_otlp_endpoint("https://localhost:4317")
+            .set_otlp_protocol(OtlpProtocol::Grpc);
+
+        assert!(builder.build::<NativeCapabilities>().is_err());
+        for _ in 0..50 {
+            if info.calls() > 0 {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert_eq!(info.calls(), 0);
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn build_with_grpc_protocol_no_endpoint_uses_agent_path() {
+        let mut builder = TraceExporterBuilder::default();
+        builder.set_otlp_protocol(OtlpProtocol::Grpc);
+        let exporter = builder.build::<NativeCapabilities>().unwrap();
+        assert!(exporter.otlp.is_none());
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn build_with_grpc_https_endpoint_rejected() {
+        let mut builder = TraceExporterBuilder::default();
+        builder
+            .set_otlp_endpoint("https://localhost:4317")
+            .set_otlp_protocol(OtlpProtocol::Grpc);
+        assert!(builder.build::<NativeCapabilities>().is_err());
+    }
+
     #[cfg_attr(miri, ignore)]
     #[test]
     fn test_build_with_v1_starts_inactive() {
diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs
index 7e522a4d3a..9b2cf4d6f6 100644
--- a/libdd-data-pipeline/src/trace_exporter/mod.rs
+++ b/libdd-data-pipeline/src/trace_exporter/mod.rs
@@ -20,8 +20,12 @@ use self::trace_serializer::TraceSerializer;
 use crate::agent_info::ResponseObserver;
 use crate::agentless::exporter::send_agentless_traces_with_observer;
 use crate::agentless::AgentlessTraceConfig;
+#[cfg(not(target_arch = "wasm32"))]
+use crate::otlp::exporter::OTLP_RETRY_DELAY_MS;
 use crate::otlp::exporter::{send_otlp_http_with_observer, OTLP_MAX_RETRIES};
 use crate::otlp::{map_traces_to_otlp, OtlpResourceInfo, OtlpTraceConfig};
+#[cfg(not(target_arch = "wasm32"))]
+use crate::otlp::{send_otlp_traces_grpc, GrpcExportError, OtlpGrpcTransport};
 #[cfg(feature = "telemetry")]
 use crate::telemetry::{SendPayloadTelemetry, TelemetryClient};
 use crate::trace_exporter::agent_response::{
@@ -57,11 +61,14 @@ use libdd_trace_utils::send_with_retry::{
 };
 use libdd_trace_utils::span::{v04::Span, TraceData};
 use libdd_trace_utils::trace_utils::TracerHeaderTags;
+#[cfg(all(feature = "telemetry", not(target_arch = "wasm32")))]
+use prost::Message;
 use std::io;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Once};
 use std::time::Duration;
 use std::{borrow::Borrow, str::FromStr};
+#[cfg(not(target_arch = "wasm32"))]
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
 
@@ -69,6 +76,36 @@ const INFO_ENDPOINT: &str = "/info";
 const V04_TRACES_ENDPOINT: &str = "/v0.4/traces";
 const V05_TRACES_ENDPOINT: &str = "/v0.5/traces";
 const V1_TRACES_ENDPOINT: &str = "/v1.0/traces";
+#[cfg(not(target_arch = "wasm32"))]
+const OTLP_GRPC_MAX_RETRY_DELAY: Duration = Duration::from_secs(30);
+#[cfg(not(target_arch = "wasm32"))]
+const OTLP_GRPC_MAX_JITTER: Duration = Duration::from_millis(100);
+
+#[cfg(not(target_arch = "wasm32"))]
+fn grpc_retry_delay(
+    attempt: u32,
+    retry_after: Option<Duration>,
+    jitter: Duration,
+) -> Option<Duration> {
+    let initial_delay = match retry_after {
+        Some(delay) if delay > OTLP_GRPC_MAX_RETRY_DELAY => return None,
+        Some(delay) if !delay.is_zero() => delay,
+        _ => Duration::from_millis(OTLP_RETRY_DELAY_MS),
+    };
+    let multiplier = 2u32.saturating_pow(attempt.saturating_sub(1));
+    Some(
+        initial_delay
+            .saturating_mul(multiplier)
+            .saturating_add(jitter)
+            .min(OTLP_GRPC_MAX_RETRY_DELAY),
+    )
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+fn grpc_retry_jitter() -> Duration {
+    let max_millis = u64::try_from(OTLP_GRPC_MAX_JITTER.as_millis()).unwrap_or(u64::MAX);
+    Duration::from_millis(rand::random::<u64>() % max_millis + 1)
+}
 
 #[derive(Clone, Copy)]
 struct PayloadCounts {
@@ -163,6 +200,13 @@ fn add_path(url: &Uri, path: &str) -> Uri {
 
 pub use libdd_trace_utils::tracer_metadata::TracerMetadata;
 
+#[derive(Debug)]
+pub(crate) enum OtlpExportMode {
+    Http(OtlpTraceConfig),
+    #[cfg(not(target_arch = "wasm32"))]
+    Grpc(OtlpGrpcTransport),
+}
+
 /// Handles for the background workers owned by a [`TraceExporter`].
 #[derive(Debug)]
 pub(crate) struct TraceExporterWorkers {
@@ -243,8 +287,10 @@ pub struct TraceExporter<
     capabilities: C,
     workers: TraceExporterWorkers,
     agent_payload_response_version: Option<AgentResponsePayloadVersion>,
-    /// When set, traces are exported via OTLP HTTP/JSON instead of the Datadog agent.
-    otlp_config: Option<OtlpTraceConfig>,
+    /// When set, traces are exported via OTLP instead of the Datadog agent.
+    otlp: Option<OtlpExportMode>,
+    /// OTLP Resource attributes derived from tracer metadata.
+    otlp_resource_info: OtlpResourceInfo,
     /// When set, APM trace spans are exported directly to the Datadog HTTP intake (agentless)
     /// instead of via the Datadog Agent
     agentless_config: Option<AgentlessTraceConfig>,
@@ -304,6 +350,37 @@ impl<
         }
     }
 
+    #[cfg(all(feature = "telemetry", not(target_arch = "wasm32")))]
+    fn emit_grpc_result(
+        &self,
+        result: &Result<(), TraceExporterError>,
+        attempts: u32,
+        bytes: usize,
+        counts: PayloadCounts,
+    ) {
+        let retry_result = match result {
+            Ok(()) => Ok((http::Response::new(Bytes::new()), attempts)),
+            Err(TraceExporterError::Request(error)) => {
+                let mut response = http::Response::new(Bytes::new());
+                *response.status_mut() = error.status();
+                Err(SendWithRetryError::Http(response, attempts))
+            }
+            Err(TraceExporterError::Io(error)) if error.kind() == std::io::ErrorKind::TimedOut => {
+                Err(SendWithRetryError::Timeout(attempts))
+            }
+            Err(TraceExporterError::Network(error))
+                if matches!(
+                    error.kind(),
+                    crate::trace_exporter::error::NetworkErrorKind::TimedOut
+                ) =>
+            {
+                Err(SendWithRetryError::Timeout(attempts))
+            }
+            Err(_) => Err(SendWithRetryError::ResponseBody(attempts)),
+        };
+        self.emit_retry_result(&retry_result, bytes, counts);
+    }
+
     /// Stop the background workers owned by this exporter.
     ///
     /// Sync facade over [`Self::shutdown_async`]; panics inside an existing tokio context.
@@ -684,33 +761,27 @@ impl<
     ) -> Result<AgentResponse, TraceExporterError> {
         #[cfg(feature = "telemetry")]
         let counts = PayloadCounts::from_traces(&traces);
-        let resource_info = {
-            let mut r = OtlpResourceInfo::default();
-            r.service = self.metadata.service.clone();
-            r.env = self.metadata.env.clone();
-            r.app_version = self.metadata.app_version.clone();
-            r.language = self.metadata.language.clone();
-            r.tracer_version = self.metadata.tracer_version.clone();
-            r.runtime_id = self.metadata.runtime_id.clone();
-            r.client_computed_stats =
-                self.metadata.client_computed_stats || self.otlp_stats_enabled;
-            r.instrumentation_scope_name = config.instrumentation_scope_name.clone();
-            r.instrumentation_scope_version = config.instrumentation_scope_version.clone();
-            r
-        };
-        // Single prost OTLP IR; the configured protocol encodes the same request to its wire
-        // format (JSON or protobuf). OTel-semantics gating (omit DD-specific attrs) happens in
-        // the mapper.
-        let request =
-            map_traces_to_otlp(traces, &resource_info, config.otel_trace_semantics_enabled);
-        let body = config.protocol.encode(&request).map_err(|e| {
-            error!("OTLP serialization error: {e}");
-            #[cfg(feature = "telemetry")]
-            self.emit_serialization_drop(counts);
-            TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!(
-                "failed to encode OTLP request: {e}"
-            )))
-        })?;
+        let request = map_traces_to_otlp(
+            traces,
+            &self.otlp_resource_info,
+            config.otel_trace_semantics_enabled,
+        );
+        let body = config
+            .protocol
+            .encode(&request)
+            .ok_or_else(|| {
+                TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(
+                    "OTLP gRPC protocol cannot be encoded on the HTTP export path".to_string(),
+                ))
+            })?
+            .map_err(|e| {
+                error!("OTLP serialization error: {e}");
+                #[cfg(feature = "telemetry")]
+                self.emit_serialization_drop(counts);
+                TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!(
+                    "failed to encode OTLP request: {e}"
+                )))
+            })?;
         // Also set the header: resource attributes survive Collector hops, headers don't.
         let effective_config;
         let config_to_use = if self.metadata.client_computed_stats || self.otlp_stats_enabled {
@@ -734,7 +805,11 @@ impl<
             &config_to_use.headers,
             config_to_use.timeout,
             self.endpoint.test_token.as_deref(),
-            config_to_use.protocol.content_type(),
+            config_to_use.protocol.content_type().ok_or_else(|| {
+                TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(
+                    "OTLP gRPC protocol cannot be sent over the HTTP export path".to_string(),
+                ))
+            })?,
             body,
             OTLP_MAX_RETRIES,
             |_result| {
@@ -747,6 +822,53 @@ impl<
         Ok(AgentResponse::Unchanged)
     }
 
+    #[cfg(not(target_arch = "wasm32"))]
+    async fn send_otlp_grpc_inner<T: TraceData>(
+        &self,
+        traces: Vec<Vec<Span<T>>>,
+        transport: &OtlpGrpcTransport,
+    ) -> Result<AgentResponse, TraceExporterError> {
+        #[cfg(feature = "telemetry")]
+        let counts = PayloadCounts::from_traces(&traces);
+        let request = Arc::new(map_traces_to_otlp(
+            traces,
+            &self.otlp_resource_info,
+            transport.otel_trace_semantics_enabled,
+        ));
+        #[cfg(feature = "telemetry")]
+        let payload_len = request.encoded_len() + 5;
+        let test_token = self.endpoint.test_token.as_deref();
+        let mut attempt: u32 = 1;
+        let result = loop {
+            match send_otlp_traces_grpc(
+                transport,
+                test_token,
+                self.metadata.client_computed_stats || self.otlp_stats_enabled,
+                request.clone(),
+            )
+            .await
+            {
+                Ok(()) => break Ok(()),
+                Err(GrpcExportError::Retryable { error, retry_after }) => {
+                    if attempt > OTLP_MAX_RETRIES {
+                        break Err(error);
+                    }
+                    let Some(delay) = grpc_retry_delay(attempt, retry_after, grpc_retry_jitter())
+                    else {
+                        break Err(error);
+                    };
+                    self.capabilities.sleep(delay).await;
+                    attempt += 1;
+                }
+                Err(GrpcExportError::NonRetryable(error)) => break Err(error),
+            }
+        };
+        #[cfg(feature = "telemetry")]
+        self.emit_grpc_result(&result, attempt, payload_len, counts);
+        result?;
+        Ok(AgentResponse::Unchanged)
+    }
+
     /// Send traces payload to agent with retry and telemetry reporting
     async fn send_traces_with_telemetry(
         &self,
@@ -842,12 +964,18 @@ impl<
         // OTLP path: send sampled traces via OTLP when an OTLP endpoint is configured.
         // Unlike the agent path, there is no downstream agent to drop unsampled traces,
         // so drop_chunks is always called here regardless of whether stats are enabled.
-        if let Some(config) = self.otlp_config.as_ref() {
+        if let Some(otlp) = &self.otlp {
             libdd_trace_utils::span::trace_utils::drop_chunks(&mut traces);
             if traces.is_empty() {
                 return Ok(AgentResponse::Unchanged);
             }
-            return self.send_otlp_traces_inner(traces, config).await;
+            return match otlp {
+                OtlpExportMode::Http(config) => self.send_otlp_traces_inner(traces, config).await,
+                #[cfg(not(target_arch = "wasm32"))]
+                OtlpExportMode::Grpc(transport) => {
+                    self.send_otlp_grpc_inner(traces, transport).await
+                }
+            };
         }
 
         // Snapshot the effective format once so the serializer and the URL agree even if
@@ -1144,6 +1272,42 @@ mod tests {
     use libdd_trace_utils::span::v04::SpanBytes;
     use std::net;
 
+    #[test]
+    fn grpc_retry_delay_applies_backoff_and_cap() {
+        let retry_after = Duration::new(2, 250_000_000);
+
+        assert_eq!(
+            grpc_retry_delay(1, Some(retry_after), Duration::ZERO),
+            Some(retry_after)
+        );
+        assert_eq!(
+            grpc_retry_delay(2, Some(retry_after), Duration::ZERO),
+            Some(Duration::new(4, 500_000_000))
+        );
+        assert_eq!(
+            grpc_retry_delay(3, None, Duration::ZERO),
+            Some(Duration::from_millis(OTLP_RETRY_DELAY_MS * 4))
+        );
+        assert_eq!(
+            grpc_retry_delay(3, Some(Duration::from_secs(20)), Duration::ZERO),
+            Some(Duration::from_secs(30))
+        );
+        assert_eq!(
+            grpc_retry_delay(1, Some(Duration::ZERO), Duration::ZERO),
+            Some(Duration::from_millis(OTLP_RETRY_DELAY_MS))
+        );
+        assert_eq!(
+            grpc_retry_delay(1, Some(Duration::from_secs(31)), Duration::ZERO),
+            None
+        );
+    }
+
+    #[test]
+    fn grpc_retry_delay_adds_bounded_jitter() {
+        let delay = grpc_retry_delay(1, None, Duration::from_millis(50));
+        assert_eq!(delay, Some(Duration::from_millis(150)));
+    }
+
     #[test]
     fn test_from_tracer_tags_to_tracer_header_tags() {
         let tracer_tags = TracerMetadata {
diff --git a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs
new file mode 100644
index 0000000000..1151f0233c
--- /dev/null
+++ b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs
@@ -0,0 +1,380 @@
+// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+#[cfg(all(test, not(target_arch = "wasm32")))]
+mod grpc_export_tests {
+    use bytes::Bytes;
+    use h2::server;
+    use libdd_capabilities_impl::NativeCapabilities;
+    #[cfg(feature = "telemetry")]
+    use libdd_data_pipeline::trace_exporter::TelemetryConfig;
+    use libdd_data_pipeline::{trace_exporter::TraceExporterBuilder, OtlpProtocol};
+    use libdd_shared_runtime::{ForkSafeRuntime, SharedRuntime};
+    use libdd_trace_protobuf::opentelemetry::proto::{
+        collector::trace::v1::{ExportTraceServiceRequest, ExportTraceServiceResponse},
+        common::v1::any_value::Value,
+    };
+    use libdd_trace_utils::test_utils::create_test_json_span;
+    use prost::Message;
+    #[cfg(feature = "telemetry")]
+    use regex::Regex;
+    use serde_json::json;
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        mpsc, Arc,
+    };
+    use std::time::Duration;
+    use tokio::net::TcpListener;
+    use tokio::sync::oneshot;
+    use tokio::task::JoinSet;
+
+    struct ReceivedExport {
+        path: String,
+        user_agent: Option<String>,
+        client_computed_stats: Option<String>,
+        request: ExportTraceServiceRequest,
+    }
+
+    async fn run_grpc_test_server(
+        listener: TcpListener,
+        req_tx: mpsc::Sender<ReceivedExport>,
+        failures_remaining: Arc<AtomicUsize>,
+        mut shutdown: oneshot::Receiver<()>,
+    ) {
+        let mut connections = JoinSet::new();
+        loop {
+            tokio::select! {
+                _ = &mut shutdown => break,
+                accepted = listener.accept() => {
+                    let Ok((socket, _)) = accepted else { return };
+                    let connection_req_tx = req_tx.clone();
+                    let connection_failures_remaining = failures_remaining.clone();
+                    connections.spawn(async move {
+                        let Ok(mut connection) = server::handshake(socket).await else {
+                            return;
+                        };
+                        let mut handlers = JoinSet::new();
+                        while let Some(result) = connection.accept().await {
+                            if let Ok((request, respond)) = result {
+                                handlers.spawn(handle_export_stream(
+                                    request,
+                                    respond,
+                                    connection_req_tx.clone(),
+                                    connection_failures_remaining.clone(),
+                                ));
+                            }
+                        }
+                        while let Some(result) = handlers.join_next().await {
+                            result.expect("gRPC request handler failed");
+                        }
+                    });
+                }
+            }
+        }
+        connections.abort_all();
+        while let Some(result) = connections.join_next().await {
+            if let Err(error) = result {
+                assert!(error.is_cancelled(), "gRPC connection task failed: {error}");
+            }
+        }
+    }
+
+    async fn handle_export_stream(
+        request: http::Request<h2::RecvStream>,
+        mut respond: h2::server::SendResponse<Bytes>,
+        req_tx: mpsc::Sender<ReceivedExport>,
+        failures_remaining: Arc<AtomicUsize>,
+    ) {
+        let path = request.uri().path().to_string();
+        let user_agent = request
+            .headers()
+            .get("user-agent")
+            .and_then(|value| value.to_str().ok())
+            .map(str::to_owned);
+        let client_computed_stats = request
+            .headers()
+            .get("datadog-client-computed-stats")
+            .and_then(|value| value.to_str().ok())
+            .map(str::to_owned);
+        let mut body = request.into_body();
+        let mut frame_data: Vec<u8> = Vec::new();
+        while let Some(chunk) = body.data().await {
+            let Ok(chunk) = chunk else { return };
+            let len = chunk.len();
+            frame_data.extend_from_slice(&chunk);
+            body.flow_control().release_capacity(len).ok();
+        }
+
+        let decoded = if frame_data.len() > 5 {
+            ExportTraceServiceRequest::decode(&frame_data[5..]).ok()
+        } else {
+            None
+        };
+        if let Some(req) = decoded {
+            let _ = req_tx.send(ReceivedExport {
+                path,
+                user_agent,
+                client_computed_stats,
+                request: req,
+            });
+        }
+
+        let mut remaining = failures_remaining.load(Ordering::Relaxed);
+        let should_fail = loop {
+            let Some(next) = remaining.checked_sub(1) else {
+                break false;
+            };
+            match failures_remaining.compare_exchange_weak(
+                remaining,
+                next,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            ) {
+                Ok(_) => break true,
+                Err(current) => remaining = current,
+            }
+        };
+
+        let response = http::Response::builder()
+            .status(200)
+            .header("content-type", "application/grpc")
+            .body(())
+            .unwrap();
+        let Ok(mut send_stream) = respond.send_response(response, false) else {
+            return;
+        };
+        if !should_fail {
+            let response_proto = ExportTraceServiceResponse::default();
+            let proto_bytes = response_proto.encode_to_vec();
+            let mut frame = Vec::with_capacity(5 + proto_bytes.len());
+            frame.push(0u8);
+            frame.extend_from_slice(
+                &u32::try_from(proto_bytes.len())
+                    .expect("protobuf response exceeds the gRPC frame length")
+                    .to_be_bytes(),
+            );
+            frame.extend_from_slice(&proto_bytes);
+            let _ = send_stream.send_data(Bytes::from(frame), false);
+        }
+
+        let mut trailers = http::HeaderMap::new();
+        trailers.insert(
+            "grpc-status",
+            if should_fail { "14" } else { "0" }.parse().unwrap(),
+        );
+        let _ = send_stream.send_trailers(trailers);
+    }
+
+    fn resource_attribute<'a>(
+        request: &'a ExportTraceServiceRequest,
+        key: &str,
+    ) -> Option<&'a str> {
+        request
+            .resource_spans
+            .first()
+            .and_then(|rs| rs.resource.as_ref())
+            .and_then(|r| {
+                r.attributes.iter().find_map(|kv| {
+                    if kv.key == key {
+                        kv.value
+                            .as_ref()
+                            .and_then(|v| v.value.as_ref())
+                            .and_then(|v| match v {
+                                Value::StringValue(s) => Some(s.as_str()),
+                                _ => None,
+                            })
+                    } else {
+                        None
+                    }
+                })
+            })
+    }
+
+    fn run_grpc_export_end_to_end_and_survives_shared_runtime_restart<'scope, 'env: 'scope>(
+        scope: &'scope std::thread::Scope<'scope, 'env>,
+    ) {
+        let (port_tx, port_rx) = mpsc::channel::<u16>();
+        let (req_tx, req_rx) = mpsc::channel::<ReceivedExport>();
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+
+        let server = scope.spawn(move || {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            rt.block_on(async move {
+                let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+                port_tx.send(listener.local_addr().unwrap().port()).unwrap();
+                run_grpc_test_server(listener, req_tx, Arc::new(AtomicUsize::new(0)), shutdown_rx)
+                    .await;
+            });
+        });
+
+        let port = port_rx
+            .recv_timeout(Duration::from_secs(10))
+            .expect("server did not bind within 10s");
+        let endpoint = format!("http://127.0.0.1:{port}/otel/");
+        let expected_path = "/otel/opentelemetry.proto.collector.trace.v1.TraceService/Export";
+
+        let shared_runtime = Arc::new(ForkSafeRuntime::new().expect("build shared runtime"));
+
+        let mut builder = TraceExporterBuilder::default();
+        builder
+            .set_shared_runtime(shared_runtime.clone())
+            .set_otlp_endpoint(&endpoint)
+            .set_otlp_protocol(OtlpProtocol::Grpc)
+            .set_connection_timeout(Some(30_000))
+            .set_language("test-lang")
+            .set_tracer_version("1.0")
+            .set_env("grpc-test-env")
+            .set_service("grpc-test-svc")
+            .set_client_computed_stats();
+
+        let exporter = builder
+            .build::<NativeCapabilities>()
+            .expect("build exporter");
+
+        let mut span = create_test_json_span(1234, 12342, 12341, 1, false);
+        span["service"] = json!("grpc-test-svc");
+        span["name"] = json!("grpc_span");
+        let data = rmp_serde::to_vec_named(&vec![vec![span]]).unwrap();
+
+        exporter.send(data.as_ref()).expect("initial send ok");
+        let initial = req_rx
+            .recv_timeout(Duration::from_secs(10))
+            .expect("server did not receive the initial request");
+        assert_eq!(initial.path, expected_path);
+        assert_eq!(
+            initial.user_agent.as_deref(),
+            Some(libdd_trace_utils::send_with_retry::TRACE_EXPORTER_USER_AGENT)
+        );
+        assert!(
+            !initial.request.resource_spans.is_empty(),
+            "expected at least one ResourceSpans"
+        );
+        assert_eq!(
+            resource_attribute(&initial.request, "service.name"),
+            Some("grpc-test-svc"),
+            "service.name attribute not found or wrong value"
+        );
+        assert_eq!(initial.client_computed_stats.as_deref(), Some("yes"));
+        assert_eq!(
+            resource_attribute(&initial.request, "_dd.stats_computed"),
+            Some("true")
+        );
+
+        shared_runtime.before_fork();
+        shared_runtime
+            .after_fork_parent()
+            .expect("restart shared runtime");
+
+        exporter.send(data.as_ref()).expect("post-restart send ok");
+        let after_restart = req_rx
+            .recv_timeout(Duration::from_secs(10))
+            .expect("server did not receive the post-restart request");
+        assert_eq!(after_restart.path, expected_path);
+        assert_eq!(
+            resource_attribute(&after_restart.request, "service.name"),
+            Some("grpc-test-svc"),
+            "service.name attribute not found or wrong value after runtime restart"
+        );
+        shutdown_tx.send(()).expect("server stopped early");
+        server.join().expect("server thread failed");
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn grpc_export_end_to_end_and_survives_shared_runtime_restart() {
+        std::thread::scope(|scope| {
+            run_grpc_export_end_to_end_and_survives_shared_runtime_restart(scope)
+        });
+    }
+
+    #[cfg(feature = "telemetry")]
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn grpc_retry_emits_native_trace_telemetry() {
+        let telemetry_server = httpmock::MockServer::start();
+        let requests_metric =
+            Regex::new(r#""metric":"trace_api.requests","points":\[\[\d+,2\.0\]\]"#).unwrap();
+        let metrics_endpoint = telemetry_server.mock(|when, then| {
+            when.method(httpmock::Method::POST)
+                .path("/telemetry/proxy/api/v2/apmtelemetry")
+                .is_true(move |request| {
+                    String::from_utf8(request.body_vec())
+                        .is_ok_and(|body| requests_metric.is_match(&body))
+                })
+                .body_includes("\"metric\":\"spans_enqueued_for_serialization\"")
+                .body_includes("\"metric\":\"trace_chunks_sent\"");
+            then.status(200)
+                .header("content-type", "application/json")
+                .body("");
+        });
+
+        std::thread::scope(|scope| {
+            let (port_tx, port_rx) = mpsc::channel::<u16>();
+            let (req_tx, req_rx) = mpsc::channel::<ReceivedExport>();
+            let (shutdown_tx, shutdown_rx) = oneshot::channel();
+            let server = scope.spawn(move || {
+                let rt = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .unwrap();
+                rt.block_on(async move {
+                    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+                    port_tx.send(listener.local_addr().unwrap().port()).unwrap();
+                    run_grpc_test_server(
+                        listener,
+                        req_tx,
+                        Arc::new(AtomicUsize::new(1)),
+                        shutdown_rx,
+                    )
+                    .await;
+                });
+            });
+
+            let port = port_rx
+                .recv_timeout(Duration::from_secs(10))
+                .expect("server did not bind within 10s");
+            let shared_runtime = Arc::new(ForkSafeRuntime::new().expect("build shared runtime"));
+            let mut builder = TraceExporterBuilder::default();
+            builder
+                .set_shared_runtime(shared_runtime)
+                .set_url(&telemetry_server.url("/"))
+                .set_otlp_endpoint(&format!("http://127.0.0.1:{port}"))
+                .set_otlp_protocol(OtlpProtocol::Grpc)
+                .set_language("test-lang")
+                .set_language_version("1.0")
+                .set_language_interpreter("test")
+                .set_tracer_version("1.0")
+                .set_service("grpc-test-svc")
+                .enable_telemetry(TelemetryConfig {
+                    heartbeat: 100,
+                    ..Default::default()
+                });
+            let exporter = builder
+                .build::<NativeCapabilities>()
+                .expect("build exporter");
+
+            let span = create_test_json_span(1234, 12342, 12341, 1, false);
+            let data = rmp_serde::to_vec_named(&vec![vec![span]]).unwrap();
+            exporter.send(data.as_ref()).expect("send traces");
+            for _ in 0..2 {
+                req_rx
+                    .recv_timeout(Duration::from_secs(10))
+                    .expect("server did not receive request");
+            }
+
+            let deadline = std::time::Instant::now() + Duration::from_secs(10);
+            while metrics_endpoint.calls() == 0 && std::time::Instant::now() < deadline {
+                std::thread::sleep(Duration::from_millis(100));
+            }
+            metrics_endpoint.assert_calls(1);
+
+            exporter
+                .shutdown(Some(Duration::from_secs(5)))
+                .expect("shutdown exporter");
+            shutdown_tx.send(()).expect("server stopped early");
+            server.join().expect("server thread failed");
+        });
+    }
+}
diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs
index 4cb3b98f1d..b14ea8af7f 100644
--- a/libdd-trace-utils/src/send_with_retry/mod.rs
+++ b/libdd-trace-utils/src/send_with_retry/mod.rs
@@ -22,6 +22,9 @@ pub type Attempts = u32;
 
 pub type SendWithRetryResult = Result<(http::Response<Bytes>, Attempts), SendWithRetryError>;
 
+/// User-agent sent by the trace exporter.
+pub const TRACE_EXPORTER_USER_AGENT: &str = concat!("Tracer/", env!("CARGO_PKG_VERSION"));
+
 /// All errors contain the number of attempts after which the final error was returned
 #[derive(Debug)]
 pub enum SendWithRetryError {
@@ -156,8 +159,7 @@ pub async fn send_with_retry_and_size<C: HttpClientCapability + SleepCapability>
         let mut builder = http::Request::builder()
             .method(http::Method::POST)
             .uri(target.url.clone());
-        builder =
-            target.set_standard_headers(builder, concat!("Tracer/", env!("CARGO_PKG_VERSION")));
+        builder = target.set_standard_headers(builder, TRACE_EXPORTER_USER_AGENT);
         for (key, value) in headers {
             builder = builder.header(key, value);
         }