fix(frontend): mint internal request ids on the server for HTTP and gRPC

docs/fern/pages/use-cases/reinforcement-learning/operations-and-simulation.md

@@ -30,7 +30,7 @@ X-Dynamo-Session-ID: rollout-run42-sample7
 X-Dynamo-Parent-Session-ID: rollout-run42-parent2
 ```
 
-Use `x-request-id` for cross-component logs and distributed traces. Native SGLang `/generate` gives a body `rid` precedence over that header, so use one stable `rid` or omit it.
+Use `x-request-id` for cross-component logs and distributed traces. Dynamo always generates its own internal request ID, so concurrent requests can reuse the same client value. Client-supplied IDs are correlation values only: native SGLang `/generate` returns the body `rid` (or `x-request-id` when `rid` is omitted) in the response, and `/inference/v1/generate` returns `x-request-id` (or the body `request_id`) as the response `request_id`.
 
 Dynamo does not define typed RL rollout, trainer-step, or policy-version fields. Keep those values in the framework and join them through opaque request or session identity. Session identity supports correlation; it does not enable affinity unless the router is configured for it and does not validate policy freshness.
 

lib/llm/src/http/service/generate.rs

@@ -76,7 +76,12 @@ fn generate_metric_error_type(status: StatusCode) -> ErrorType {
 
 #[derive(Debug)]
 struct GenerateRequestContext {
+    /// Server-minted id for the request context, routing, and logs. Never client-chosen:
+    /// in-flight request state is keyed on it.
     request_id: String,
+    /// Client-facing id echoed in the response body and the `vllm_tito` envelope, following
+    /// vLLM's `x-request-id` > body `request_id` precedence. Correlation only.
+    public_request_id: String,
     data_parallel_rank: Option<u32>,
 }
 
@@ -130,19 +135,21 @@ fn resolve_generate_request_context(
     headers: &HeaderMap,
     body_request_id: Option<&str>,
 ) -> GenerateRequestContext {
-    let request_id = headers
+    let request_id = get_or_create_request_id(headers);
+    let public_request_id = headers
         .get(X_REQUEST_ID_HEADER)
         .and_then(|value| value.to_str().ok())
         .map(ToOwned::to_owned)
         .or_else(|| body_request_id.map(ToOwned::to_owned))
-        .unwrap_or_else(|| get_or_create_request_id(headers));
+        .unwrap_or_else(|| request_id.clone());
     let data_parallel_rank = headers
         .get(X_DATA_PARALLEL_RANK_HEADER)
         .and_then(|value| value.to_str().ok())
         .and_then(|value| value.trim().parse().ok());
 
     GenerateRequestContext {
         request_id,
+        public_request_id,
         data_parallel_rank,
     }
 }
@@ -601,7 +608,7 @@ fn preprocessed_from_generate(
     request: GenerateRequest,
     model: &str,
     data_parallel_rank: Option<u32>,
-    request_id: &str,
+    public_request_id: &str,
     routing_metadata: GenerateRoutingMetadata,
 ) -> anyhow::Result<PreprocessedRequest> {
     let tracker = Arc::new(RequestTracker::new());
@@ -610,7 +617,7 @@ fn preprocessed_from_generate(
         request,
         model,
         data_parallel_rank,
-        request_id,
+        public_request_id,
         routing_metadata,
         tracker,
     )
@@ -620,7 +627,7 @@ fn preprocessed_from_generate_with_tracker(
     request: GenerateRequest,
     model: &str,
     data_parallel_rank: Option<u32>,
-    request_id: &str,
+    public_request_id: &str,
     routing_metadata: GenerateRoutingMetadata,
     tracker: Arc<RequestTracker>,
 ) -> anyhow::Result<PreprocessedRequest> {
@@ -661,7 +668,7 @@ fn preprocessed_from_generate_with_tracker(
             }
         }
     };
-    let vllm_tito = serde_json::to_value(VllmTitoEnvelope::new(&request, request_id))?;
+    let vllm_tito = serde_json::to_value(VllmTitoEnvelope::new(&request, public_request_id))?;
     let mut extra_args = serde_json::Map::new();
     extra_args.insert("vllm_tito".to_string(), vllm_tito);
     if let Some(kv_transfer_params) = request.kv_transfer_params.as_ref() {
@@ -973,7 +980,7 @@ async fn handler_generate(
         request,
         &model,
         request_context.data_parallel_rank,
-        &request_context.request_id,
+        &request_context.public_request_id,
         routing_metadata,
         tracker,
     ) {
@@ -1018,6 +1025,7 @@ async fn handler_generate(
             engine,
             context,
             request_id,
+            request_context.public_request_id,
             state.clone(),
             response_options,
             metric_lifecycle,
@@ -1041,6 +1049,7 @@ async fn generate_dispatch(
     engine: crate::types::openai::generate::GenerateStreamingEngine,
     context: Context<PreprocessedRequest>,
     request_id: String,
+    public_request_id: String,
     state: Arc<service_v2::State>,
     response_options: GenerateResponseOptions,
     metric_lifecycle: GenerateMetricLifecycle,
@@ -1126,7 +1135,7 @@ async fn generate_dispatch(
         request_context.as_ref(),
         GenerateResponse::from_annotated_stream_with_options(
             stream,
-            request_id.clone(),
+            public_request_id,
             response_options,
         ),
     )
@@ -1756,6 +1765,140 @@ pub(crate) mod tests {
         );
     }
 
+    /// Echoes each request's tokens back, and holds every request until `barrier` releases them
+    /// together so both are in flight at once. Records what the worker saw for each request.
+    struct ConcurrentEchoEngine {
+        barrier: tokio::sync::Barrier,
+        seen: Mutex<Vec<(String, Option<String>)>>,
+    }
+
+    #[async_trait::async_trait]
+    impl AsyncEngine<SingleIn<PreprocessedRequest>, ManyOut<Annotated<LLMEngineOutput>>, Error>
+        for ConcurrentEchoEngine
+    {
+        async fn generate(
+            &self,
+            request: SingleIn<PreprocessedRequest>,
+        ) -> Result<ManyOut<Annotated<LLMEngineOutput>>, Error> {
+            let envelope_request_id = request
+                .extra_args
+                .as_ref()
+                .and_then(|extra| extra["vllm_tito"]["request_id"].as_str())
+                .map(str::to_string);
+            self.seen
+                .lock()
+                .unwrap()
+                .push((request.id().to_string(), envelope_request_id));
+            self.barrier.wait().await;
+            let stream = futures::stream::iter([Annotated::from_data(LLMEngineOutput {
+                token_ids: request.token_ids.to_vec(),
+                index: Some(0),
+                finish_reason: Some(crate::protocols::common::FinishReason::Stop),
+                ..Default::default()
+            })]);
+            Ok(ResponseStream::new(Box::pin(stream), request.context()))
+        }
+    }
+
+    /// Serve `engine` through a frontend source/sink pipeline, like the discovery watcher
+    /// does, so responses are matched to callers by request id.
+    pub(crate) fn engine_behind_frontend(
+        engine: crate::types::openai::generate::GenerateStreamingEngine,
+    ) -> crate::types::openai::generate::GenerateStreamingEngine {
+        use dynamo_runtime::pipeline::{ServiceBackend, Source};
+        let frontend = dynamo_runtime::pipeline::ServiceFrontend::<
+            SingleIn<PreprocessedRequest>,
+            ManyOut<Annotated<LLMEngineOutput>>,
+        >::new();
+        frontend
+            .link(ServiceBackend::from_engine(engine))
+            .unwrap()
+            .link_terminal(frontend)
+            .unwrap()
+    }
+
+    async fn concurrent_generate_with_shared_client_id(
+        headers: HeaderMap,
+        body_request_id: Option<&str>,
+        expected_public_id: &str,
+    ) {
+        const MODEL: &str = "generate-concurrent-model";
+        let engine = Arc::new(ConcurrentEchoEngine {
+            barrier: tokio::sync::Barrier::new(2),
+            seen: Mutex::new(Vec::new()),
+        });
+        let service = HttpService::builder().build().unwrap();
+        let state = service.state_clone();
+        state
+            .manager()
+            .add_generate_model(
+                MODEL,
+                "generate-concurrent-test",
+                engine_behind_frontend(engine.clone()),
+            )
+            .unwrap();
+
+        let send = |token_ids: Vec<u32>| {
+            let mut body = serde_json::json!({
+                "token_ids": token_ids,
+                "sampling_params": {},
+                "model": MODEL
+            });
+            if let Some(id) = body_request_id {
+                body["request_id"] = serde_json::json!(id);
+            }
+            let request = serde_json::from_value(body).unwrap();
+            handler_generate(State(state.clone()), headers.clone(), Json(request))
+        };
+        let (first, second) = tokio::time::timeout(
+            Duration::from_secs(10),
+            futures::future::join(send(vec![1, 2, 3]), send(vec![7, 8])),
+        )
+        .await
+        .expect("concurrent generate requests must not hang");
+
+        for (response, expected_tokens) in [(first, vec![1, 2, 3]), (second, vec![7, 8])] {
+            assert_eq!(response.status(), StatusCode::OK);
+            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+                .await
+                .unwrap();
+            let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
+            assert_eq!(body["request_id"], expected_public_id);
+            assert_eq!(
+                body["choices"][0]["token_ids"],
+                serde_json::json!(expected_tokens)
+            );
+        }
+
+        let seen = engine.seen.lock().unwrap();
+        assert_eq!(seen.len(), 2);
+        assert_ne!(
+            seen[0].0, seen[1].0,
+            "each request needs its own context id"
+        );
+        for (context_id, envelope_request_id) in seen.iter() {
+            assert_ne!(context_id, expected_public_id);
+            assert_eq!(envelope_request_id.as_deref(), Some(expected_public_id));
+        }
+    }
+
+    #[tokio::test]
+    async fn concurrent_generate_requests_sharing_x_request_id_get_their_own_output() {
+        let mut headers = HeaderMap::new();
+        headers.insert(X_REQUEST_ID_HEADER, "1".parse().unwrap());
+        concurrent_generate_with_shared_client_id(headers, None, "1").await;
+    }
+
+    #[tokio::test]
+    async fn concurrent_generate_requests_sharing_body_request_id_get_their_own_output() {
+        concurrent_generate_with_shared_client_id(
+            HeaderMap::new(),
+            Some("shared-body-id"),
+            "shared-body-id",
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn generate_early_failures_are_counted() {
         let service = HttpService::builder().build().unwrap();
@@ -2501,7 +2644,8 @@ pub(crate) mod tests {
 
         let context = resolve_generate_request_context(&headers, Some("body-request"));
 
-        assert_eq!(context.request_id, "header-request");
+        assert_eq!(context.public_request_id, "header-request");
+        assert!(uuid::Uuid::parse_str(&context.request_id).is_ok());
         assert_eq!(context.data_parallel_rank, Some(3));
     }
 
@@ -2512,8 +2656,33 @@ pub(crate) mod tests {
 
         let context = resolve_generate_request_context(&headers, Some("body-request"));
 
-        assert_eq!(context.request_id, "body-request");
+        assert_eq!(context.public_request_id, "body-request");
+        assert!(uuid::Uuid::parse_str(&context.request_id).is_ok());
         assert_eq!(context.data_parallel_rank, None);
+
+        let context = resolve_generate_request_context(&HeaderMap::new(), None);
+        assert_eq!(context.public_request_id, context.request_id);
+    }
+
+    #[test]
+    fn generate_request_context_mints_internal_id_for_every_request() {
+        let mut headers = HeaderMap::new();
+        headers.insert(X_REQUEST_ID_HEADER, "shared".parse().unwrap());
+        headers.insert(
+            "x-dynamo-request-id",
+            "11111111-2222-3333-4444-555555555555".parse().unwrap(),
+        );
+
+        let first = resolve_generate_request_context(&headers, Some("shared"));
+        let second = resolve_generate_request_context(&headers, Some("shared"));
+
+        assert_eq!(first.public_request_id, "shared");
+        assert_eq!(second.public_request_id, "shared");
+        assert_ne!(first.request_id, second.request_id);
+        for context in [&first, &second] {
+            assert_ne!(context.request_id, "shared");
+            assert_ne!(context.request_id, "11111111-2222-3333-4444-555555555555");
+        }
     }
 
     #[test]
@@ -2584,6 +2753,7 @@ pub(crate) mod tests {
         generate_dispatch(
             engine,
             context,
+            request_id.clone(),
             request_id,
             state,
             response_options,

lib/llm/src/http/service/openai.rs

@@ -1031,43 +1031,25 @@ pub async fn smart_json_error_middleware(request: Request<Body>, next: Next) ->
 
 /// Return the request ID for the current request.
 ///
-/// The canonical request ID is set by `make_inference_request_span()` and stored
+/// The canonical request ID is minted by `make_inference_request_span()` and stored
 /// in the `DistributedTraceContext` via `DistributedTraceIdLayer`. This function
-/// retrieves it, falling back to a validated `x-dynamo-request-id` header value
-/// (deprecated, DEP #7812) or a new UUID.
+/// retrieves it, or generates a new UUID when no request span is active.
 ///
-/// **Deprecation (DEP #7812):** The `x-dynamo-request-id` header is deprecated.
-/// Clients should rely on server-generated request IDs instead of supplying their own.
+/// Client-supplied ids never become the internal request id: in-flight request state
+/// is keyed on this value, so two requests sharing a client id would collide. Clients
+/// correlate requests with `x-request-id` instead.
+///
+/// **Deprecation (DEP #7812):** the `x-dynamo-request-id` and `request-id` headers are
+/// ignored on HTTP requests; a warning is logged when either is present.
 pub(super) fn get_or_create_request_id(headers: &HeaderMap) -> String {
-    // Validate x-dynamo-request-id header if present, warn on invalid values.
-    // DEP #7812: x-dynamo-request-id is deprecated — clients should rely on
-    // server-generated request IDs instead of supplying their own.
-    let validated_header = if let Some(raw) = headers.get(DYNAMO_REQUEST_ID_HEADER) {
-        tracing::warn!(
-            "{} header is deprecated (DEP #7812); server-generated request IDs should be used instead",
-            DYNAMO_REQUEST_ID_HEADER
-        );
-        match raw.to_str() {
-            Err(_) => {
-                tracing::warn!(
-                    "{} header must be a valid UTF-8 string",
-                    DYNAMO_REQUEST_ID_HEADER
-                );
-                None
-            }
-            Ok(s) if uuid::Uuid::parse_str(s).is_err() => {
-                tracing::warn!(
-                    "{} header must be a valid UUID, got: {}",
-                    DYNAMO_REQUEST_ID_HEADER,
-                    s
-                );
-                None
-            }
-            Ok(s) => Some(s.to_string()),
+    for header in [DYNAMO_REQUEST_ID_HEADER, "request-id"] {
+        if headers.contains_key(header) {
+            tracing::warn!(
+                header,
+                "deprecated request id header ignored (DEP #7812); the server generates the request ID, use x-request-id for client correlation"
+            );
         }
-    } else {
-        None
-    };
+    }
 
     // Prefer trace context (set by make_inference_request_span via DistributedTraceIdLayer)
     if let Some(trace_context) = get_distributed_tracing_context()
@@ -1076,8 +1058,7 @@ pub(super) fn get_or_create_request_id(headers: &HeaderMap) -> String {
         return request_id;
     }
 
-    // Fallback: use validated header for backwards compat, or generate new UUID
-    validated_header.unwrap_or_else(|| uuid::Uuid::new_v4().to_string())
+    uuid::Uuid::new_v4().to_string()
 }
 
 pub(super) fn context_from_headers<T: Send + Sync + 'static>(