feat(sglang): upload sidecar metadata with OpenDAL

lib/llm/src/preprocessor.rs

@@ -9025,7 +9025,8 @@ mod tests {
                 "cache_salt": "step_7",
                 "extra_fields": ["completion_token_ids"],
                 "metadata_upload": {
-                    "url": "s3://bucket/root/rollouts"
+                    "url": "s3://bucket/root/rollouts",
+                    "fallback_url": "file:///var/tmp/rollouts"
                 }
             }
         }))
@@ -9041,7 +9042,8 @@ mod tests {
         assert_eq!(
             extra_args["nvext"]["metadata_upload"],
             serde_json::json!({
-                "url": "s3://bucket/root/rollouts"
+                "url": "s3://bucket/root/rollouts",
+                "fallback_url": "file:///var/tmp/rollouts"
             })
         );
         assert_eq!(extra_args["sampling_options"]["detokenize"], false);

lib/llm/src/protocols/common/extensions.rs

@@ -46,18 +46,41 @@ pub struct RouterParams {
 pub struct MetadataUpload {
     #[serde(deserialize_with = "deserialize_metadata_upload_url")]
     pub url: String,
+
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        deserialize_with = "deserialize_metadata_upload_fallback_url"
+    )]
+    pub fallback_url: Option<String>,
 }
 
 fn deserialize_metadata_upload_url<'de, D>(deserializer: D) -> Result<String, D::Error>
 where
     D: serde::Deserializer<'de>,
 {
     let url = String::deserialize(deserializer)?;
+    normalize_metadata_upload_url(url, "metadata_upload.url")
+}
+
+fn deserialize_metadata_upload_fallback_url<'de, D>(
+    deserializer: D,
+) -> Result<Option<String>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    Option::<String>::deserialize(deserializer)?
+        .map(|url| normalize_metadata_upload_url(url, "metadata_upload.fallback_url"))
+        .transpose()
+}
+
+fn normalize_metadata_upload_url<E>(url: String, field: &str) -> Result<String, E>
+where
+    E: serde::de::Error,
+{
     let url = url.trim();
     if url.is_empty() {
-        return Err(serde::de::Error::custom(
-            "metadata_upload.url must not be empty",
-        ));
+        return Err(E::custom(format!("{field} must not be empty")));
     }
     Ok(url.to_string())
 }
@@ -992,16 +1015,21 @@ mod tests {
     }
 
     #[test]
-    fn metadata_upload_parses_url() {
+    fn metadata_upload_parses_urls() {
         let nvext: NvExt = serde_json::from_value(serde_json::json!({
             "metadata_upload": {
-                "url": " s3://bucket/root/rollouts "
+                "url": " s3://bucket/root/rollouts ",
+                "fallback_url": " file:///var/tmp/rollouts "
             }
         }))
         .unwrap();
 
         let upload = nvext.metadata_upload.as_ref().unwrap();
         assert_eq!(upload.url, "s3://bucket/root/rollouts");
+        assert_eq!(
+            upload.fallback_url.as_deref(),
+            Some("file:///var/tmp/rollouts")
+        );
         assert!(!NvExtResponseFieldSelection::from_nvext(Some(&nvext)).engine_data);
 
         assert!(
@@ -1018,6 +1046,15 @@ mod tests {
             }))
             .is_err()
         );
+        assert!(
+            serde_json::from_value::<NvExt>(serde_json::json!({
+                "metadata_upload": {
+                    "url": "s3://bucket/root/rollouts",
+                    "fallback_url": ""
+                }
+            }))
+            .is_err()
+        );
         assert!(
             serde_json::from_value::<NvExt>(serde_json::json!({
                 "metadata_upload": {

lib/sidecar/sglang/Cargo.toml

@@ -29,13 +29,23 @@ async-stream = { workspace = true }
 async-trait = { workspace = true }
 clap = { version = "4", features = ["derive", "env"] }
 futures = { workspace = true }
+opendal = { version = "0.59.1", default-features = false, features = [
+    "http-transport-reqwest",
+    "services-azblob",
+    "services-fs",
+    "services-gcs",
+    "services-s3",
+] }
 rand = { workspace = true }
 reqwest = { workspace = true }
+rmp-serde = { workspace = true }
+serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
+zstd = "0.13"
 
 prost = { workspace = true }
 tonic = { workspace = true }
@@ -45,3 +55,4 @@ tonic-build = { workspace = true }
 
 [dev-dependencies]
 dynamo-backend-common = { workspace = true, features = ["testing"] }
+tempfile = { workspace = true }

lib/sidecar/sglang/README.md

@@ -28,6 +28,57 @@ follow-up.
 
 Use `DYN_SIDECAR_GRPC_ENDPOINT` instead of `--grpc-endpoint` when the endpoint is provided through the environment.
 
+## Upload response metadata
+
+Start the sidecar with `--enable-rl` or `DYN_ENABLE_RL=true` to allow response
+metadata uploads. Uploads remain disabled for requests that omit
+`nvext.metadata_upload`.
+
+Set `nvext.metadata_upload.url` in an OpenAI-compatible generation request:
+
+```json
+{
+  "nvext": {
+    "metadata_upload": {
+      "url": "s3://bucket/root/rollout-1",
+      "fallback_url": "file:///var/tmp/rollout-1"
+    }
+  }
+}
+```
+
+The sidecar writes `choice_0.msgpack.zst` under this URL.
+The artifact uses the same MessagePack and Zstandard schema as the Python SGLang bindings.
+The supported URL schemes are `file://`, `s3://`, `gs://`, and `az://`.
+OpenDAL loads credentials from the standard provider configuration for the selected service.
+The optional `fallback_url` can use a different supported scheme.
+If the primary upload fails, the sidecar writes the same artifact to the fallback URL.
+
+OpenDAL selects the storage service from each URL scheme.
+The packaged sidecar includes the file, S3, Google Cloud Storage, and Azure Blob services.
+A downstream binary can register a custom OpenDAL service before it starts the sidecar.
+The custom service must use the compatible OpenDAL registry and link into that binary.
+The metadata upload code does not need a change for the custom scheme.
+The packaged executable does not load service plug-ins at run time.
+
+The sidecar uploads metadata only for aggregated and decode requests.
+It does not return the uploaded metadata inline.
+If the final upload fails, the sidecar stops the terminal response.
+Thus, the caller cannot accept a rollout without its metadata.
+
+| Frontend API or worker role | Metadata upload support |
+|---|---|
+| `POST /v1/chat/completions` | Supported |
+| `POST /v1/completions` | Supported |
+| `POST /v1/responses` | Supported |
+| Native SGLang `POST` or `PUT /generate` | Not supported; the native request adapter does not project `nvext.metadata_upload` into Dynamo request metadata |
+| Aggregated worker | Uploads final `meta_info` |
+| Disaggregated prefill worker | Does not upload |
+| Disaggregated decode worker | Uploads final `meta_info` |
+
+The sidecar supports one output choice, so each request writes
+`choice_0.msgpack.zst`.
+
 Native Dynamo `/generate` requests are forwarded opaquely to SGLang's HTTP
 endpoint using the gRPC host and the HTTP port returned by `GetServerInfo`.
 The sidecar advertises this capability only after the HTTP health probe passes

lib/sidecar/sglang/src/engine.rs

@@ -22,6 +22,7 @@ use tokio_util::sync::CancellationToken;
 
 use crate::args::Args;
 use crate::client::{self, Client, Discovery, Pool};
+use crate::metadata_upload::{MetadataUploader, grpc_metadata};
 use crate::native_http::{self, NativeHttp};
 use crate::proto as pb;
 use crate::protocol::{
@@ -35,6 +36,7 @@ pub struct SglangSidecarEngine {
     disaggregation_mode: DisaggregationMode,
     bootstrap_host: Option<String>,
     bootstrap_port: Option<u16>,
+    metadata_upload_enabled: bool,
     state: OnceCell<StartedState>,
     cancel: CancellationToken,
 }
@@ -102,6 +104,7 @@ impl SglangSidecarEngine {
         );
 
         let common = args.sidecar.common;
+        let metadata_upload_enabled = common.enable_rl;
         let config = WorkerConfig {
             namespace: common.namespace,
             component: if disaggregation_mode == DisaggregationMode::Aggregated {
@@ -135,6 +138,7 @@ impl SglangSidecarEngine {
                 disaggregation_mode,
                 bootstrap_host,
                 bootstrap_port,
+                metadata_upload_enabled,
                 state: OnceCell::new(),
                 cancel: CancellationToken::new(),
             },
@@ -249,6 +253,10 @@ impl LLMEngine for SglangSidecarEngine {
             .state
             .get()
             .ok_or_else(|| client::engine_shutdown("generate called before start"))?;
+        let metadata_uploader = MetadataUploader::from_request(
+            &request,
+            self.metadata_upload_enabled && !self.disaggregation_mode.is_prefill(),
+        )?;
         if let Some(native_request) = native_http::request(
             &request,
             ctx.id(),
@@ -261,7 +269,12 @@ impl LLMEngine for SglangSidecarEngine {
                     "native SGLang Generate is unavailable because no ready incremental HTTP endpoint was discovered",
                 )
             })?;
-            return Ok(native_http.generate(native_request, ctx, self.cancel.clone()));
+            return Ok(native_http.generate(
+                native_request,
+                ctx,
+                self.cancel.clone(),
+                metadata_uploader,
+            ));
         }
         let mut grpc_client = state.pool.stream_client();
 
@@ -353,14 +366,17 @@ impl LLMEngine for SglangSidecarEngine {
                                 break;
                             }
                         };
-                        let (log_probs, top_logprobs) =
+                        let (log_probs, top_logprobs) = if metadata_uploader.is_some() {
+                            (None, None)
+                        } else {
                             match extract_logprobs(&response.meta_info, return_tokens_as_ids) {
                                 Ok(values) => values,
                                 Err(err) => {
                                     yield Err(err);
                                     break;
                                 }
-                            };
+                            }
+                        };
 
                         if is_prefill {
                             if response.finished {
@@ -395,12 +411,20 @@ impl LLMEngine for SglangSidecarEngine {
                                     break;
                                 }
                             };
-                            let engine_data = match engine_data_from_meta(&response.meta_info, true) {
-                                Ok(engine_data) => engine_data,
-                                Err(error) => {
+                            let engine_data = if let Some(uploader) = metadata_uploader.as_ref() {
+                                if let Err(error) = uploader.upload(grpc_metadata(&response.meta_info)).await {
                                     yield Err(error);
                                     break;
                                 }
+                                None
+                            } else {
+                                match engine_data_from_meta(&response.meta_info, true) {
+                                    Ok(engine_data) => engine_data,
+                                    Err(error) => {
+                                        yield Err(error);
+                                        break;
+                                    }
+                                }
                             };
                             terminal.token_ids = token_ids;
                             terminal.log_probs = log_probs;
@@ -411,11 +435,15 @@ impl LLMEngine for SglangSidecarEngine {
                         }
 
                         if !token_ids.is_empty() {
-                            let engine_data = match engine_data_from_meta(&response.meta_info, false) {
-                                Ok(engine_data) => engine_data,
-                                Err(error) => {
-                                    yield Err(error);
-                                    break;
+                            let engine_data = if metadata_uploader.is_some() {
+                                None
+                            } else {
+                                match engine_data_from_meta(&response.meta_info, false) {
+                                    Ok(engine_data) => engine_data,
+                                    Err(error) => {
+                                        yield Err(error);
+                                        break;
+                                    }
                                 }
                             };
                             yield Ok(LLMEngineOutput {

lib/sidecar/sglang/src/lib.rs

@@ -9,12 +9,13 @@
 //! role, parallelism, KV block sizing, and context length are discovered from
 //! the engine's gRPC metadata RPCs.
 //!
-//! The crate never depends on `sglang` or any engine crate — only
-//! `dynamo-backend-common`, `tonic`/`prost`, `clap`, and tokio.
+//! The crate never depends on `sglang` or another engine crate.
+//! It uses the generated protocol contract and OpenDAL storage services.
 
 pub mod args;
 pub mod client;
 pub mod engine;
+mod metadata_upload;
 mod native_http;
 
 /// Generated SGLang gRPC types, temporarily exposed for the Mocker server