feat(pylon): handle Dynamo request priority in Pylon

src/libraries/rust/stargate/crates/pylon-lib/src/bringup/upstream.rs

@@ -94,7 +94,7 @@ pub(super) async fn send_completion_request(
                 request_id: request_id.clone(),
                 routing_key: None,
                 model_id: model_id.to_string(),
-                priority: 0,
+                priority: None,
                 input_tokens: u64::try_from(input_tokens).unwrap_or(u64::MAX),
                 accepted_at: std::time::Instant::now(),
             },

src/libraries/rust/stargate/crates/pylon-lib/src/lib.rs

@@ -45,9 +45,9 @@ pub use model_lifecycle::{
 };
 pub use queue_admission::PylonQueueMismatchRetryConfig;
 pub use quic_http_tunnel::{
-    DEFAULT_MAX_SSE_BUFFER_BYTES, PylonRetryConfig, QuicHttpTunnelConfig, QuicHttpTunnelHandle,
-    ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, TunnelError, TunnelForwardingConfig,
-    start_quic_http_tunnel, start_reverse_quic_tunnel,
+    DEFAULT_MAX_SSE_BUFFER_BYTES, DEFAULT_PRIORITY_CEILING, PylonRetryConfig, QuicHttpTunnelConfig,
+    QuicHttpTunnelHandle, ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, TunnelError,
+    TunnelForwardingConfig, UpstreamBackend, start_quic_http_tunnel, start_reverse_quic_tunnel,
 };
 pub use registration::{
     ClientError, InferenceServerRegistrationClient, InferenceServerRegistrationConfig,

src/libraries/rust/stargate/crates/pylon-lib/src/queue_admission.rs

@@ -304,7 +304,7 @@ impl LiveRequestState {
                         })
                         .filter(|request| request.generation == *generation);
                     model.queue_estimate_ms_for_priority_excluding(
-                        required.priority,
+                        required.queue_priority(),
                         excluded_request,
                     )
                 })
@@ -346,7 +346,7 @@ impl LiveRequestState {
         let request_id = required.request_id.clone();
         let request = TrackedPromptRequest {
             generation,
-            priority: required.priority,
+            priority: required.queue_priority(),
             input_tokens: required.input_tokens,
             phase: TrackedPromptPhase::Pending,
             active_chat_output_tps: None,
@@ -842,7 +842,7 @@ mod tests {
             request_id: request_id.to_string(),
             routing_key: None,
             model_id: model_id.to_string(),
-            priority,
+            priority: Some(priority),
             input_tokens,
             accepted_at: Instant::now(),
         }

src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel.rs

@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod backend;
 mod core;
 mod endpoint;
 mod http3;
@@ -23,6 +24,7 @@ mod server;
 mod tests;
 mod webtransport;
 
+pub use backend::{DEFAULT_PRIORITY_CEILING, UpstreamBackend};
 pub use core::{DEFAULT_MAX_SSE_BUFFER_BYTES, PylonRetryConfig, TunnelForwardingConfig};
 pub use endpoint::TunnelError;
 pub use reverse::{ReverseQuicTunnelConfig, ReverseQuicTunnelHandle, start_reverse_quic_tunnel};

src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/backend.rs

@@ -0,0 +1,108 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Upstream inference-server dialects.
+//!
+//! Pylon speaks the platform tunnel contract upward and translates it into
+//! the dialect of the engine it fronts. All engine-specific header names and
+//! encodings live in this module.
+
+use std::fmt;
+use std::str::FromStr;
+
+/// Which engine dialect pylon speaks to its local upstream. Future engines
+/// add a variant and a submodule here, never a new CLI flag.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum UpstreamBackend {
+    /// Forward requests unchanged; derive nothing. Inbound engine priority
+    /// headers are still stripped.
+    Passthrough,
+    /// Derive the engine priority headers from `x-priority`.
+    #[default]
+    Dynamo,
+}
+
+impl fmt::Display for UpstreamBackend {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Passthrough => "passthrough",
+            Self::Dynamo => "dynamo",
+        })
+    }
+}
+
+impl FromStr for UpstreamBackend {
+    type Err = String;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value.trim().to_ascii_lowercase().as_str() {
+            "passthrough" => Ok(Self::Passthrough),
+            "dynamo" => Ok(Self::Dynamo),
+            other => Err(format!(
+                "unknown upstream backend {other:?}; expected \"passthrough\" or \"dynamo\""
+            )),
+        }
+    }
+}
+
+/// Default priority band ceiling; see [`dynamo::request_priority`].
+pub const DEFAULT_PRIORITY_CEILING: u32 = 3600;
+
+pub(crate) mod dynamo {
+    use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
+
+    /// Engine priority headers pylon derives; the names stay out of the
+    /// shared tunnel contract because only pylon speaks them.
+    pub(crate) const HEADER_REQUEST_PRIORITY: &str = "x-dynamo-request-priority";
+    pub(crate) const HEADER_REQUEST_STRICT_PRIORITY: &str = "x-dynamo-request-strict-priority";
+
+    /// Denylist of engine headers pylon owns: inbound values are stripped in
+    /// every backend mode so pylon stays their only writer. Scoped to the
+    /// priority headers for now; other engine headers are tracked separately.
+    const STRIPPED_REQUEST_HEADERS: [&str; 2] =
+        [HEADER_REQUEST_PRIORITY, HEADER_REQUEST_STRICT_PRIORITY];
+
+    pub(crate) fn is_stripped_engine_header(name: &HeaderName) -> bool {
+        STRIPPED_REQUEST_HEADERS.contains(&name.as_str())
+    }
+
+    /// Map the platform rank (lower wins, absent = unconfigured) to the
+    /// engine value (higher wins, read as seconds of queue head start):
+    /// `max(0, ceiling - rank)`, with absent as the lowest value. The head
+    /// start is bounded so prioritized traffic cannot starve the rest.
+    pub(crate) fn request_priority(priority: Option<u32>, ceiling: u32) -> i32 {
+        let ceiling = ceiling.min(i32::MAX as u32);
+        let rank = priority.unwrap_or(ceiling).min(ceiling);
+        (ceiling - rank) as i32
+    }
+
+    /// Emit both priority headers on every inference request.
+    pub(crate) fn apply_priority_headers(
+        priority: Option<u32>,
+        ceiling: u32,
+        upstream_headers: &mut HeaderMap,
+    ) -> i32 {
+        let dynamo_priority = request_priority(priority, ceiling);
+        upstream_headers.insert(
+            HeaderName::from_static(HEADER_REQUEST_PRIORITY),
+            HeaderValue::from(dynamo_priority),
+        );
+        upstream_headers.insert(
+            HeaderName::from_static(HEADER_REQUEST_STRICT_PRIORITY),
+            HeaderValue::from_static("0"),
+        );
+        dynamo_priority
+    }
+}

src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/core.rs

@@ -37,6 +37,7 @@ use tokio_util::{sync::CancellationToken, task::TaskTracker};
 use tracing::{Instrument, Span, field};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
+use super::backend::{self, DEFAULT_PRIORITY_CEILING, UpstreamBackend};
 use crate::output_token_parser::{OutputTokenParser, OutputTokenProgress};
 use crate::queue_admission::{
     PylonQueueMismatchRetryConfig, QueueAdmissionDecision, QueueTrackedRequestGuard,
@@ -105,6 +106,10 @@ pub struct TunnelForwardingConfig {
     pub request_quality_monitor: RequestQualityMonitorConfig,
     pub retry: PylonRetryConfig,
     pub queue_mismatch_retry: PylonQueueMismatchRetryConfig,
+    /// Engine dialect spoken to the local upstream; see [`UpstreamBackend`].
+    pub upstream_backend: UpstreamBackend,
+    /// Priority band ceiling; see [`backend::dynamo::request_priority`].
+    pub priority_ceiling: u32,
     pub metrics: Option<Arc<PylonMetrics>>,
     #[cfg(test)]
     pub webtransport_stream_header_wait_tx: Option<flume::Sender<()>>,
@@ -121,6 +126,8 @@ impl Default for TunnelForwardingConfig {
             request_quality_monitor: RequestQualityMonitorConfig::default(),
             retry: PylonRetryConfig::default(),
             queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(),
+            upstream_backend: UpstreamBackend::default(),
+            priority_ceiling: DEFAULT_PRIORITY_CEILING,
             metrics: None,
             #[cfg(test)]
             webtransport_stream_header_wait_tx: None,
@@ -141,6 +148,8 @@ pub(super) struct TunnelServerApp {
     pub(super) request_quality_monitor: RequestQualityMonitorConfig,
     pub(super) retry: PylonRetryConfig,
     pub(super) queue_mismatch_retry: PylonQueueMismatchRetryConfig,
+    pub(super) upstream_backend: UpstreamBackend,
+    pub(super) priority_ceiling: u32,
     pub(super) metrics: Option<Arc<PylonMetrics>>,
     #[cfg(test)]
     pub(super) webtransport_stream_header_wait_tx: Option<flume::Sender<()>>,
@@ -164,6 +173,8 @@ impl TunnelServerApp {
             request_quality_monitor: forwarding.request_quality_monitor,
             retry: forwarding.retry,
             queue_mismatch_retry: forwarding.queue_mismatch_retry,
+            upstream_backend: forwarding.upstream_backend,
+            priority_ceiling: forwarding.priority_ceiling,
             metrics: forwarding.metrics,
             #[cfg(test)]
             webtransport_stream_header_wait_tx: forwarding.webtransport_stream_header_wait_tx,
@@ -654,13 +665,17 @@ pub(super) async fn forward_tunnel_request(
         }
     }
 
+    let priority = lifecycle
+        .as_ref()
+        .and_then(|lifecycle| lifecycle.required.priority);
     let response = match send_upstream_request(
         app,
         method,
         &path_and_query,
         &request_headers,
         body_bytes,
-        !health_request,
+        health_request,
+        priority,
     )
     .await
     {
@@ -699,9 +714,10 @@ async fn send_upstream_request(
     path_and_query: &str,
     request_headers: &HeaderMap,
     body_bytes: Vec<u8>,
-    traced: bool,
+    health_request: bool,
+    priority: Option<u32>,
 ) -> Result<Response, UpstreamRequestError> {
-    let span = if traced {
+    let span = if !health_request {
         let span = tracing::info_span!(
             "pylon_upstream_http_request",
             otel_parent = field::Empty,
@@ -710,6 +726,8 @@ async fn send_upstream_request(
             inference_server.id = %app.inference_server_id,
             upstream.status = field::Empty,
             upstream.error = field::Empty,
+            priority = field::Empty,
+            dynamo.request_priority = field::Empty,
         );
         let _ = span.set_parent(pylon_upstream_parent_context(request_headers));
         if let Some(otel_parent) = otel_parent_from_headers(request_headers) {
@@ -725,7 +743,18 @@ async fn send_upstream_request(
             upstream_headers.append(name, value.clone());
         }
     }
-    if traced {
+    if !health_request {
+        if let Some(priority) = priority {
+            span.record("priority", priority);
+        }
+        if app.upstream_backend == UpstreamBackend::Dynamo {
+            let dynamo_priority = backend::dynamo::apply_priority_headers(
+                priority,
+                app.priority_ceiling,
+                &mut upstream_headers,
+            );
+            span.record("dynamo.request_priority", dynamo_priority);
+        }
         inject_trace_context(&mut upstream_headers, &span.context());
     }
     let send = async {
@@ -1058,6 +1087,7 @@ pub(super) fn join_base_path(base: &str, path_and_query: &str) -> Result<url::Ur
 
 pub(super) fn should_forward_header(name: &HeaderName, retry: &PylonRetryConfig) -> bool {
     !is_tunnel_control_header(name, retry)
+        && !backend::dynamo::is_stripped_engine_header(name)
         && !matches!(
             name.as_str(),
             "host" | "x-method" | "x-path" | HEADER_STARGATE_EXPECTED_QUEUE_MS

src/libraries/rust/stargate/crates/pylon-lib/src/quic_http_tunnel/raw_quic.rs

@@ -171,6 +171,8 @@ mod tests {
             request_quality_monitor: RequestQualityMonitorConfig::default(),
             retry: PylonRetryConfig::default(),
             queue_mismatch_retry: PylonQueueMismatchRetryConfig::default(),
+            upstream_backend: crate::quic_http_tunnel::UpstreamBackend::default(),
+            priority_ceiling: crate::quic_http_tunnel::DEFAULT_PRIORITY_CEILING,
             metrics: None,
             #[cfg(test)]
             webtransport_stream_header_wait_tx: None,