feat(sidecar): direct-gRPC dispatch from router to stock engine containers (v2)

Cargo.toml

@@ -29,6 +29,8 @@ members = [
     "lib/sidecar/sglang",
     "lib/mocker/servers/sglang",
     "lib/sidecar/trtllm",
+    "lib/sidecar/frontend",
+    "lib/sidecar/register",
     "lib/bindings/c",
     "lib/bindings/python/codegen",
     "deploy/inference-gateway/ext-proc",

lib/backend-common/src/worker.rs

@@ -833,13 +833,15 @@ impl Worker {
         self.state = LifecycleState::Stopped;
     }
 
-    /// Drive the serve loop and the shutdown orchestrator. Returns when
-    /// either the serve loop exits or `shutdown` is cancelled.
-    async fn serve_with_orchestrator(
+    /// Model-registration prologue shared by both orchestrators: resolve the
+    /// model/worker type, build the local model, hand the engine its endpoint
+    /// (a fatal handoff, done before discovery so a failure leaves nothing
+    /// published), and attach the discovery Model record. Callers add the
+    /// transport-specific tail (request-plane ingress vs. direct-gRPC register).
+    async fn register_model(
         &mut self,
         engine_config: &EngineConfig,
-        endpoint: dynamo_runtime::component::Endpoint,
-        shutdown: CancellationToken,
+        endpoint: &dynamo_runtime::component::Endpoint,
     ) -> Result<(), DynamoError> {
         let model_type = resolve_model_type(&self.config)?;
         let (worker_type, needs) = resolve_worker_type_and_needs(&self.config);
@@ -848,18 +850,11 @@ impl Worker {
             build_local_model(&self.config, engine_config, self.engine.is_raw()).await?;
         tracing::debug!("local model built");
 
-        // Hand the engine its serving endpoint before registering the model
-        // with discovery. on_endpoint_ready is a fatal handoff: doing it first
-        // means a failure leaves nothing published, so there is no stale
-        // discovery entry to reclaim. Engines that publish their own discovery
-        // records (e.g. vLLM dynamic LoRA) stash the endpoint here, and this
-        // still runs before `register_engine_controls`, so `/engine/*` cannot
-        // fire before the engine has the endpoint.
         self.engine.on_endpoint_ready(endpoint.clone()).await?;
 
         local_model
             .attach(
-                &endpoint,
+                endpoint,
                 model_type,
                 self.config.model_input,
                 None,
@@ -874,6 +869,21 @@ impl Worker {
                 )
             })?;
         tracing::debug!("model registered with discovery");
+        Ok(())
+    }
+
+    /// Drive the serve loop and the shutdown orchestrator. Returns when
+    /// either the serve loop exits or `shutdown` is cancelled.
+    async fn serve_with_orchestrator(
+        &mut self,
+        engine_config: &EngineConfig,
+        endpoint: dynamo_runtime::component::Endpoint,
+        shutdown: CancellationToken,
+    ) -> Result<(), DynamoError> {
+        // on_endpoint_ready (inside register_model) runs before
+        // register_engine_controls, so `/engine/*` cannot fire before the engine
+        // has its endpoint.
+        self.register_model(engine_config, &endpoint).await?;
 
         self.register_engine_controls(&endpoint).await?;
         self.register_engine_updates(&endpoint).await?;

lib/bindings/python/rust/lib.rs

@@ -770,6 +770,7 @@ impl TransportType {
         match &self.inner {
             rs::component::TransportType::Nats(_) => "nats_tcp",
             rs::component::TransportType::Tcp(_) => "tcp",
+            rs::component::TransportType::Grpc(_) => "grpc",
         }
     }
 

lib/llm/src/discovery.rs

@@ -30,6 +30,12 @@ pub use endpoint_card::wait_for_endpoint_model_card;
 mod watcher;
 pub use watcher::{ModelUpdate, ModelWatcher};
 
+pub mod direct_dispatch;
+pub use direct_dispatch::{
+    DIRECT_BACKEND_KEY, DirectDispatchProvider, LlmStreamingDispatch, direct_dispatch_provider,
+    register_direct_dispatch_provider,
+};
+
 mod worker_monitor;
 pub use worker_monitor::{
     KvWorkerMonitor, LoadThresholdConfig, WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL, WorkerLoadState,

lib/llm/src/discovery/direct_dispatch.rs

@@ -0,0 +1,71 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Composition-root registry for direct-gRPC transport dispatch providers.
+//!
+//! When a worker registers with `runtime_data["direct_backend"] = "<name>"`, the
+//! frontend dispatches inference straight to that worker's native engine gRPC
+//! server instead of over the Dynamo request plane — while `PushRouter` keeps
+//! instance selection, occupancy, fault detection, and migration (only the
+//! transport below the seam changes; see [`dynamo_runtime::pipeline::StreamingDispatch`]).
+//!
+//! The engine-specific translation lives OUTSIDE `dynamo-llm` (e.g. in the
+//! TensorRT-LLM sidecar crate). It is injected here via a provider registered at
+//! a composition root (a frontend binary / the Python bindings) before the
+//! frontend runs. This keeps `dynamo-llm` engine-agnostic — a Rust
+//! dependency-inversion boundary, not a dynamic-plugin ABI.
+
+use std::collections::HashMap;
+use std::sync::{Arc, OnceLock};
+
+use async_trait::async_trait;
+use dynamo_runtime::pipeline::StreamingDispatch;
+use dynamo_runtime::protocols::annotated::Annotated;
+use parking_lot::RwLock;
+
+use crate::model_card::ModelDeploymentCard;
+use crate::protocols::common::llm_backend::{LLMEngineOutput, PreprocessedRequest};
+
+/// `runtime_data` key a direct-backend worker sets to name its dispatch provider
+/// (e.g. `"trtllm"`). The registrar writes it; the watcher reads it to select a
+/// [`DirectDispatchProvider`]. Single source shared across the write/read/lookup
+/// sites so they can't drift.
+pub const DIRECT_BACKEND_KEY: &str = "direct_backend";
+
+/// The transport-seam engine `PushRouter` dispatches through for an LLM model:
+/// typed `PreprocessedRequest` in, `Annotated<LLMEngineOutput>` out.
+pub type LlmStreamingDispatch =
+    Arc<dyn StreamingDispatch<PreprocessedRequest, Annotated<LLMEngineOutput>>>;
+
+/// Builds a direct-gRPC transport dispatch for a model whose worker advertises
+/// `runtime_data["direct_backend"] == self.backend()`.
+#[async_trait]
+pub trait DirectDispatchProvider: Send + Sync {
+    /// The `direct_backend` name this provider handles (e.g. `"trtllm"`).
+    fn backend(&self) -> &str;
+
+    /// Build the transport dispatch for one model. Engine parameters (context
+    /// length, etc.) come from the model card; per-instance gRPC addresses are
+    /// resolved per request from the routed `AddressedRequest`.
+    async fn build(&self, card: &ModelDeploymentCard) -> anyhow::Result<LlmStreamingDispatch>;
+}
+
+type Registry = RwLock<HashMap<String, Arc<dyn DirectDispatchProvider>>>;
+
+fn registry() -> &'static Registry {
+    static REGISTRY: OnceLock<Registry> = OnceLock::new();
+    REGISTRY.get_or_init(|| RwLock::new(HashMap::new()))
+}
+
+/// Register a provider at a composition root. Last registration for a given
+/// backend name wins. Call before running the frontend.
+pub fn register_direct_dispatch_provider(provider: Arc<dyn DirectDispatchProvider>) {
+    registry()
+        .write()
+        .insert(provider.backend().to_string(), provider);
+}
+
+/// Look up the provider registered for a `direct_backend` name, if any.
+pub fn direct_dispatch_provider(backend: &str) -> Option<Arc<dyn DirectDispatchProvider>> {
+    registry().read().get(backend).cloned()
+}

lib/llm/src/discovery/watcher.rs

@@ -1724,6 +1724,30 @@ impl ModelWatcher {
             worker_set.encoder_router = encoder_chooser.clone();
 
             let preprocessed_routing = if needs_preprocessed_routing {
+                // Direct-gRPC backend? Build its transport dispatch from the
+                // composition-root-registered provider so the router dispatches
+                // straight to the engine's gRPC server instead of the request plane.
+                let direct_dispatch = match card
+                    .runtime_config
+                    .runtime_data
+                    .get(crate::discovery::DIRECT_BACKEND_KEY)
+                    .and_then(|v| v.as_str())
+                {
+                    Some(backend) => match crate::discovery::direct_dispatch_provider(backend) {
+                        Some(provider) => Some(
+                            provider
+                                .build(card)
+                                .await
+                                .context("build direct-gRPC dispatch")?,
+                        ),
+                        None => anyhow::bail!(
+                            "model advertises direct_backend={backend:?} but no \
+                             DirectDispatchProvider is registered; the frontend was not \
+                             built/configured with that backend's provider"
+                        ),
+                    },
+                    None => None,
+                };
                 Some(
                     entrypoint::build_preprocessed_routing(
                         &client,
@@ -1735,6 +1759,7 @@ impl ModelWatcher {
                         encoder_chooser.clone(),
                         uses_multimodal_cache_routing(card),
                         router_config.session_affinity_ttl_secs,
+                        direct_dispatch,
                     )
                     .await
                     .context("build_preprocessed_routing")?,

lib/llm/src/entrypoint/input/common.rs

@@ -252,6 +252,10 @@ pub async fn build_preprocessed_routing(
     encoder_chooser: Option<Arc<EncoderRouter>>,
     enable_multimodal_cache_indexer: bool,
     session_affinity_ttl_secs: Option<u64>,
+    // When `Some`, the model is a direct-gRPC backend: swap the router's transport
+    // seam for this dispatch instead of the request-plane `AddressedPushRouter`.
+    // All selection / fault-detection / migration behavior is unchanged.
+    direct_dispatch: Option<crate::discovery::LlmStreamingDispatch>,
 ) -> anyhow::Result<PreprocessedRouting> {
     // Fail fast on an unsupported LoRA + router-mode combination BEFORE waiting for the initial
     // worker set, so a misconfiguration surfaces immediately at startup rather than after the
@@ -261,6 +265,12 @@ pub async fn build_preprocessed_routing(
         model_manager.lora_enabled(),
         session_affinity_ttl_secs.is_some(),
     )?;
+    if direct_dispatch.is_some() && matches!(router_mode, RouterMode::DeviceAwareWeighted) {
+        anyhow::bail!(
+            "direct-gRPC backends do not support DeviceAwareWeighted routing \
+             (no multimodal embedding-cache indexer on the direct path)"
+        );
+    }
     let min_initial_workers = min_initial_workers_from_env()?;
     let router_client = router_client(client, router_mode, chooser.as_ref())?;
 
@@ -288,14 +298,23 @@ pub async fn build_preprocessed_routing(
     let monitor_arc =
         worker_monitor.map(|m| Arc::new(m) as Arc<dyn dynamo_runtime::pipeline::WorkerLoadMonitor>);
 
-    let router = LlmPushRouter::from_client_with_state(
-        router_client,
-        router_mode,
-        monitor_arc,
-        embedding_cache_indexer,
-        cache_key_extractor,
-    )
-    .await?;
+    let router = match direct_dispatch {
+        // Direct-gRPC backend: keep all PushRouter behavior, swap only the
+        // final-hop transport (request plane -> gRPC).
+        Some(dispatch) => {
+            LlmPushRouter::from_client_with_dispatch(router_client, router_mode, dispatch).await?
+        }
+        None => {
+            LlmPushRouter::from_client_with_state(
+                router_client,
+                router_mode,
+                monitor_arc,
+                embedding_cache_indexer,
+                cache_key_extractor,
+            )
+            .await?
+        }
+    };
 
     // Eagerly register router request metrics so they appear as zeros even in
     // non-KV modes (Direct, Random, RoundRobin) where KvPushRouter is never created.